diff --git a/.agents/skills/create-in-app-notification/SKILL.md b/.agents/skills/create-in-app-notification/SKILL.md
new file mode 100644
index 0000000000..a781741818
--- /dev/null
+++ b/.agents/skills/create-in-app-notification/SKILL.md
@@ -0,0 +1,227 @@
+---
+name: Create In-App Notification
+description: Create or update a Baserow in-app notification for an event. Use when adding a backend `NotificationType`, wiring frontend notification rendering and routing, defining the notification target, or preventing duplicate notifications for the same event object.
+---
+
+# Create Baserow In-App Notifications
+
+Use this skill when a task is to add or update an in-app notification shown in Baserow's notification center.
+
+Do not invent a new notification architecture. This repo already has established backend and frontend patterns. Start from the closest existing notification type in the same product area: core, database, builder, automation, integration, premium, or enterprise.
+
+## First Step
+
+Before editing, identify which shape best matches the event:
+
+1. One event sends one notification to one or more explicit users.
+2. One event fans out to many users and should be grouped or queued efficiently.
+3. One event is instance-wide and should be a broadcast notification.
+4. The event should update or reuse an existing notification instead of creating another one.
+
+Then inspect the closest example before editing.
+
+Useful starting points:
+
+- Core notification types: `backend/src/baserow/core/notification_types.py`
+- Database notification types: `backend/src/baserow/contrib/database/fields/notification_types.py`
+- Premium notification types: `premium/backend/src/baserow_premium/row_comments/notification_types.py`
+- Enterprise notification types: `enterprise/backend/src/baserow_enterprise/data_scanner/notification_types.py`
+- Backend notification APIs: `backend/src/baserow/core/notifications/handler.py`
+- Backend notification base classes: `backend/src/baserow/core/notifications/registries.py`
+- Frontend base notification type: `web-frontend/modules/core/notificationTypes.js`
+
+## What A Complete Notification Usually Needs
+
+Most new notifications touch both sides:
+
+1. Backend `NotificationType` subclass and event hook.
+2. Backend registration in the relevant app `ready()` method.
+3. Frontend `NotificationType` class.
+4. Frontend content component used in the notification list.
+5. Frontend registration in the relevant `plugin.js`.
+6. Targeted backend tests, and frontend tests if the route or rendering logic is non-trivial.
+
+## Backend Pattern
+
+Follow the existing backend shape:
+
+1. Add a typed payload container, usually a dataclass, with the minimal stable fields needed by the UI and routing.
+2. Implement a `NotificationType` subclass.
+3. Add a helper like `create_notification`, `notify_*`, or `construct_notification`.
+4. Call `NotificationHandler.create_direct_notification_for_users(...)` for direct notifications.
+5. Use `NotificationHandler.construct_notification(...)` plus `UserNotificationsGrouper` when batching many notifications.
+6. Use `NotificationHandler.create_broadcast_notification(...)` only for true broadcast events.
+7. Register the notification type in the matching backend app.
+
+Common backend registration points:
+
+- `backend/src/baserow/core/apps.py`
+- `backend/src/baserow/contrib/database/apps.py`
+- `premium/backend/src/baserow_premium/apps.py`
+- `enterprise/backend/src/baserow_enterprise/apps.py`
+
+## Frontend Pattern
+
+If the notification must render inside the app, add the frontend type too:
+
+1. Create a frontend `NotificationType` subclass with the same `type` string.
+2. Return the appropriate icon component.
+3. Return a content component that renders the notification text.
+4. Implement `getRoute(notificationData)` when the notification should be clickable.
+5. Register the type in the relevant frontend `plugin.js`.
+
+Common frontend registration points:
+
+- `web-frontend/modules/core/plugin.js`
+- `web-frontend/modules/database/plugin.js`
+- `premium/web-frontend/modules/baserow_premium/plugin.js`
+- `enterprise/web-frontend/modules/baserow_enterprise/plugin.js`
+
+## Define The Target Clearly
+
+Every notification should have a clear target: what object or page the user should land on when they click it.
+
+Prefer storing stable identifiers in `notification.data`, not display-only values. Usually that means IDs plus enough names to render a readable message.
+
+Good target payload examples:
+
+- Row or field event: `database_id`, `table_id`, `row_id`, `field_id`
+- Comment event: `comment_id`, `table_id`, `row_id`
+- Workspace-scoped event: `workspace_id` or object IDs resolvable within the workspace
+- Admin or global event: IDs and query parameters needed for an admin route
+
+Use these rules:
+
+1. Include the smallest set of IDs required to reconstruct the target route.
+2. Include names only for display or email text.
+3. Keep the target stable even if labels change later.
+4. Use `workspace=None` only when the event is truly user-global or instance-global.
+5. If the backend email link should point to the same place, keep the backend and frontend route assumptions aligned.
+
+There are two target implementations to consider:
+
+1. Backend `get_web_frontend_url(...)`
+   Use this when the notification is emailed and should link into the app.
+   `EmailNotificationTypeMixin` already provides the default `/notification/<workspace_id>/<notification_id>` route when `has_web_frontend_route = True`.
+   Override it only when the target route cannot be expressed through that default flow.
+2. Frontend `getRoute(notificationData)`
+   Return the real in-app route object based on the IDs stored in `notification.data`.
+
+If the notification redirects through the generic notification route, verify the frontend route can still resolve the final location from the stored data.
+
+## Prevent Duplicate Notifications
+
+Do not blindly create a new notification every time a signal fires. First decide whether repeated events should:
+
+1. Create a new notification every time.
+2. Reuse one existing unread notification for the same object.
+3. Suppress re-creation while a tracking record still exists.
+4. Mark an existing notification as read instead of creating a new one.
+
+The repo uses several duplicate-prevention patterns already:
+
+### Pattern 1: Query for an existing active notification
+
+Use this when the event has a natural unique object, such as an invitation, comment, sync, or scan.
+
+Typical lookup shape:
+
+```python
+NotificationHandler.get_notification_by(
+    user,
+    notificationrecipient__read=False,
+    data__contains={"some_object_id": obj.id},
+)
+```
+
+Use this when you want at most one active notification per recipient and per object. If one already exists, do not create another. Depending on the behavior, you can:
+
+- return early
+- update the existing notification data
+- mark the existing notification as read as part of a follow-up action
+
+Choose `data` keys that uniquely identify the event target. If multiple objects can share the same notification type, the dedupe key must include all IDs needed to distinguish them.
+
+### Pattern 2: Persist or reuse an event-tracking row
+
+Use this when the same source content may be removed and re-added quickly, and a raw notification query is not enough.
+
+The rich text mention flow uses `RichTextFieldMention` rows to track mention existence and avoid duplicate notifications when content is briefly undone and redone. Follow that approach when the event source has lifecycle state that should outlive a single signal call.
+
+### Pattern 3: Group creation before writing recipients
+
+Use `UserNotificationsGrouper` when one operation can generate many notifications across many users. This reduces fan-out overhead and avoids ad hoc per-user creation loops.
+
+### Pattern 4: Update or mark read instead of inserting
+
+If the event resolves a prior notification, prefer updating state over inserting another notification. For example, invitation follow-up flows mark the original invitation notification as read.
+
+## Choosing A Dedupe Key
+
+A dedupe key is usually an implicit tuple made from:
+
+1. notification `type`
+2. recipient user
+3. active state, usually unread and uncleared
+4. one or more stable object IDs stored in `data`
+
+Examples:
+
+- One notification per invitation per user:
+  `type + recipient + data.invitation_id`
+- One notification per row comment mention per user:
+  `type + recipient + data.comment_id`
+- One notification per row-field mention per user:
+  `type + recipient + data.field_id + data.row_id`
+
+Do not dedupe on mutable names or message text.
+
+## Implementation Checklist
+
+When adding a new notification, verify all of these:
+
+1. The `type` string is unique and stable.
+2. The payload contains stable target IDs.
+3. The workspace is correct for permission-scoped listing.
+4. The sender is correct, or `None` if there is no meaningful sender.
+5. Duplicate creation behavior is explicit.
+6. Backend registration is present.
+7. Frontend registration is present if the notification appears in-app.
+8. The route works for the intended target.
+9. Tests cover both creation and duplicate prevention behavior.
+
+## Testing Expectations
+
+Add the narrowest backend tests that prove:
+
+1. The right recipients are selected.
+2. The notification payload contains the target IDs.
+3. The notification is workspace-scoped correctly.
+4. A duplicate event does not create an extra active notification when dedupe is required.
+5. The route-related data needed by the frontend is present.
+
+Useful existing tests:
+
+- `premium/backend/tests/baserow_premium_tests/row_comments/test_row_comments_notification_types.py`
+- `enterprise/backend/tests/baserow_enterprise_tests/data_scanner/test_data_scanner_notification_types.py`
+
+If you add custom frontend routing or rendering logic, add or update a focused frontend unit test near the notification type or component.
+
+## Search Patterns
+
+Use these searches to move quickly:
+
+- `rg -n "class .*NotificationType" backend/src premium/backend/src enterprise/backend/src`
+- `rg -n "notification_type_registry.register" backend/src premium/backend/src enterprise/backend/src`
+- `rg -n "new .*NotificationType\\(context\\)" web-frontend premium/web-frontend enterprise/web-frontend`
+- `rg -n "NotificationHandler\\.create_direct_notification_for_users|UserNotificationsGrouper|create_broadcast_notification" backend/src premium/backend/src enterprise/backend/src`
+- `rg -n "data__contains=.*_id|get_notification_by\\(" backend/src premium/backend/src enterprise/backend/src`
+
+## Guardrails
+
+- Do not create a new notification type without checking whether an existing one should be reused or updated.
+- Do not store only display text if the notification needs to link back to an object.
+- Do not dedupe on mutable fields like names or messages.
+- Do not use broadcasts for ordinary per-user events.
+- Do not skip frontend registration when the notification must render in-app.
+- Do not create duplicate unread notifications for the same object unless that is explicitly the desired product behavior.
diff --git a/.editorconfig b/.editorconfig
index b537486312..3c84e5486a 100644
--- a/.editorconfig
+++ b/.editorconfig
@@ -11,7 +11,7 @@ insert_final_newline = true
 [Makefile]
 indent_style = tab
 
-[*.{js,mjs,yml,scss,eslintrc,stylelintrc,vue,html,json,ts,prettierrc}]
+[*.{js,jsx,mjs,yml,scss,eslintrc,stylelintrc,vue,html,json,ts,tsx,prettierrc}]
 indent_size = 2
 
 [*.md]
diff --git a/AGENTS.md b/AGENTS.md
index 6b911cf6d2..903248df52 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -21,6 +21,12 @@ For direct package-manager use, backend commands run through `uv` and frontend c
 
 Python targets Python 3.14, uses 4-space indentation, and is formatted and linted with Ruff (`ruff check`, `ruff format`) with an 88-character line length. Follow existing Django app/module naming and keep new tests in `test_*.py` or `*_test.py` files. Frontend code uses ESLint, Stylelint, and Prettier; SCSS should follow BEM-style naming already used in `web-frontend/modules`.
 
+## Technology Stack
+
+Backend code uses Django, Django REST Framework, Celery, PostgreSQL, Redis, and pytest/pytest-django. Python dependencies are managed with `uv`.
+
+Frontend code uses Vue 3, Nuxt 3, Vuex, Vite, Vitest, Storybook, SCSS, ESLint, Stylelint, Prettier, and `yarn`. Render functions must use Vue 3 semantics, for example importing `h` from `vue` instead of expecting `render(h)` to receive it. JSX-bearing frontend files must use a `.jsx` or `.tsx` extension so Vite can parse them.
+
 ## Testing Guidelines
 
 Backend tests use `pytest` with `pytest-django`; frontend tests use `vitest`; browser flows live in `e2e-tests/`. Add unit tests for backend changes and targeted frontend tests for component or store behavior. 
@@ -40,6 +46,7 @@ Reusable skills live in `.agents/skills/`. Each subdirectory is a self-contained
 | `add-django-config-env-var` | Adding a new Django setting backed by an env var and propagating it to `base.py`, docker-compose files, `env-remap.mjs`, and `docs/installation/configuration.md` |
 | `write-frontend-unit-test` | Writing or fixing frontend unit tests in `web-frontend`, `premium/web-frontend`, or `enterprise/web-frontend` |
 | `create-update-service` | Creating or updating an integration type or service type in `contrib/integrations` |
+| `create-in-app-notification` | Creating or updating a Baserow in-app notification for an event, including backend and frontend registration, target routing data, and duplicate-prevention behavior |
 
 ## Security & Configuration Tips
 
diff --git a/backend/src/baserow/config/settings/base.py b/backend/src/baserow/config/settings/base.py
index ca2f614f3e..3091301277 100644
--- a/backend/src/baserow/config/settings/base.py
+++ b/backend/src/baserow/config/settings/base.py
@@ -857,6 +857,9 @@ def __setitem__(self, key, value):
 _automation_workflow_rate_limits_env = os.getenv(
     "BASEROW_AUTOMATION_WORKFLOW_RATE_LIMITS"
 )
+_automation_workflow_error_limits_env = os.getenv(
+    "BASEROW_AUTOMATION_WORKFLOW_ERROR_LIMITS"
+)
 
 if _automation_workflow_rate_limits_env is not None:
     _automation_workflow_rate_limit_values = [
@@ -894,6 +897,28 @@ def __setitem__(self, key, value):
         _legacy_workflow_rate_limit_window_seconds or 5,
     )
 )
+if _automation_workflow_error_limits_env is not None:
+    _automation_workflow_error_limit_values = [
+        int(value.strip())
+        for value in _automation_workflow_error_limits_env.split(",")
+        if value.strip()
+    ]
+else:
+    _automation_workflow_error_limit_values = [20, 300]
+
+if len(_automation_workflow_error_limit_values) % 2 != 0:
+    raise ImproperlyConfigured(
+        "BASEROW_AUTOMATION_WORKFLOW_ERROR_LIMITS must contain an even number of "
+        "comma-separated integers formatted as max_errors,window_seconds pairs."
+    )
+
+AUTOMATION_WORKFLOW_ERROR_LIMITS = tuple(
+    (
+        _automation_workflow_error_limit_values[index],
+        _automation_workflow_error_limit_values[index + 1],
+    )
+    for index in range(0, len(_automation_workflow_error_limit_values), 2)
+)
 AUTOMATION_WORKFLOW_MAX_CONSECUTIVE_ERRORS = int(
     os.getenv("BASEROW_AUTOMATION_WORKFLOW_MAX_CONSECUTIVE_ERRORS", 5)
 )
diff --git a/backend/src/baserow/contrib/automation/api/workflows/errors.py b/backend/src/baserow/contrib/automation/api/workflows/errors.py
index 85f12c1c83..5cd29a0b7e 100644
--- a/backend/src/baserow/contrib/automation/api/workflows/errors.py
+++ b/backend/src/baserow/contrib/automation/api/workflows/errors.py
@@ -12,6 +12,12 @@
     "The workflow id {e.workflow_id} does not belong to the automation.",
 )
 
+ERROR_AUTOMATION_WORKFLOW_NOTIFICATION_RECIPIENTS_INVALID = (
+    "ERROR_AUTOMATION_WORKFLOW_NOTIFICATION_RECIPIENTS_INVALID",
+    HTTP_400_BAD_REQUEST,
+    "{e}",
+)
+
 ERROR_AUTOMATION_NODE_DOES_NOT_EXIST = (
     "ERROR_AUTOMATION_NODE_DOES_NOT_EXIST",
     HTTP_404_NOT_FOUND,
diff --git a/backend/src/baserow/contrib/automation/api/workflows/serializers.py b/backend/src/baserow/contrib/automation/api/workflows/serializers.py
index 3c38b4271d..4707515688 100644
--- a/backend/src/baserow/contrib/automation/api/workflows/serializers.py
+++ b/backend/src/baserow/contrib/automation/api/workflows/serializers.py
@@ -16,6 +16,7 @@
 class AutomationWorkflowSerializer(serializers.ModelSerializer):
     published_on = serializers.SerializerMethodField()
     state = serializers.SerializerMethodField()
+    notification_recipient_ids = serializers.SerializerMethodField()
 
     class Meta:
         model = AutomationWorkflow
@@ -29,6 +30,7 @@ class Meta:
             "published_on",
             "state",
             "graph",
+            "notification_recipient_ids",
         )
         extra_kwargs = {
             "id": {"read_only": True},
@@ -47,6 +49,14 @@ def get_state(self, obj):
         published_workflow = AutomationWorkflowHandler().get_published_workflow(obj)
         return published_workflow.state if published_workflow else WorkflowState.DRAFT
 
+    @extend_schema_field(serializers.ListField(child=serializers.IntegerField()))
+    def get_notification_recipient_ids(self, obj):
+        """
+        Use the prefetched recipients.
+        """
+
+        return sorted((recipient.id for recipient in obj.notification_recipients.all()))
+
 
 class CreateAutomationWorkflowSerializer(serializers.ModelSerializer):
     class Meta:
@@ -62,10 +72,23 @@ class UpdateAutomationWorkflowSerializer(serializers.ModelSerializer):
             f"{ALLOW_TEST_RUN_MINUTES} minutes."
         ),
     )
+    notification_recipient_ids = serializers.ListField(
+        child=serializers.IntegerField(),
+        required=False,
+        help_text=(
+            "The user IDs of the workspace members that should receive "
+            "notifications related to this workflow."
+        ),
+    )
 
     class Meta:
         model = AutomationWorkflow
-        fields = ("name", "allow_test_run", "state")
+        fields = (
+            "name",
+            "allow_test_run",
+            "state",
+            "notification_recipient_ids",
+        )
         extra_kwargs = {
             "name": {"required": False},
         }
diff --git a/backend/src/baserow/contrib/automation/api/workflows/views.py b/backend/src/baserow/contrib/automation/api/workflows/views.py
index 609c14acd5..cc6201526b 100644
--- a/backend/src/baserow/contrib/automation/api/workflows/views.py
+++ b/backend/src/baserow/contrib/automation/api/workflows/views.py
@@ -20,6 +20,7 @@
 from baserow.contrib.automation.api.workflows.errors import (
     ERROR_AUTOMATION_WORKFLOW_DOES_NOT_EXIST,
     ERROR_AUTOMATION_WORKFLOW_NOT_IN_AUTOMATION,
+    ERROR_AUTOMATION_WORKFLOW_NOTIFICATION_RECIPIENTS_INVALID,
 )
 from baserow.contrib.automation.api.workflows.serializers import (
     AutomationWorkflowHistorySerializer,
@@ -37,6 +38,7 @@
 )
 from baserow.contrib.automation.workflows.exceptions import (
     AutomationWorkflowDoesNotExist,
+    AutomationWorkflowNotificationRecipientsInvalid,
     AutomationWorkflowNotInAutomation,
 )
 from baserow.contrib.automation.workflows.job_types import (
@@ -72,6 +74,7 @@ class AutomationWorkflowsView(APIView):
             200: AutomationWorkflowSerializer,
             400: get_error_schema(
                 [
+                    "ERROR_AUTOMATION_WORKFLOW_NOTIFICATION_RECIPIENTS_INVALID",
                     "ERROR_REQUEST_BODY_VALIDATION",
                 ]
             ),
@@ -82,6 +85,9 @@ class AutomationWorkflowsView(APIView):
     @map_exceptions(
         {
             ApplicationDoesNotExist: ERROR_APPLICATION_DOES_NOT_EXIST,
+            AutomationWorkflowNotificationRecipientsInvalid: (
+                ERROR_AUTOMATION_WORKFLOW_NOTIFICATION_RECIPIENTS_INVALID
+            ),
         }
     )
     @validate_body(CreateAutomationWorkflowSerializer, return_validated=True)
@@ -147,6 +153,7 @@ def get(self, request, workflow_id: int):
             200: AutomationWorkflowSerializer,
             400: get_error_schema(
                 [
+                    "ERROR_AUTOMATION_WORKFLOW_NOTIFICATION_RECIPIENTS_INVALID",
                     "ERROR_REQUEST_BODY_VALIDATION",
                 ]
             ),
@@ -163,6 +170,9 @@ def get(self, request, workflow_id: int):
         {
             ApplicationDoesNotExist: ERROR_APPLICATION_DOES_NOT_EXIST,
             AutomationWorkflowDoesNotExist: ERROR_AUTOMATION_WORKFLOW_DOES_NOT_EXIST,
+            AutomationWorkflowNotificationRecipientsInvalid: (
+                ERROR_AUTOMATION_WORKFLOW_NOTIFICATION_RECIPIENTS_INVALID
+            ),
         }
     )
     @validate_body(UpdateAutomationWorkflowSerializer, return_validated=True)
diff --git a/backend/src/baserow/contrib/automation/application_types.py b/backend/src/baserow/contrib/automation/application_types.py
index 2030f706dd..a4c9c8ea34 100644
--- a/backend/src/baserow/contrib/automation/application_types.py
+++ b/backend/src/baserow/contrib/automation/application_types.py
@@ -108,6 +108,7 @@ def export_serialized(
         serialized_workflows = [
             handler.export_workflow(
                 w,
+                import_export_config=import_export_config,
                 files_zip=files_zip,
                 storage=storage,
                 cache=self.cache,
@@ -208,6 +209,7 @@ def import_serialized(
             storage,
             progress.create_child_builder(represents_progress=100),
         )
+
         automation = application.specific
 
         if not serialized_integrations:
@@ -261,8 +263,15 @@ def fetch_workflows_to_serialize(
             instance = self.enhance_queryset(base_queryset).first()
             return instance and list(instance.workflows.all()) or []
 
+    def _get_workflows_queryset(self) -> QuerySet[AutomationWorkflow]:
+        return AutomationWorkflow.objects.select_related(
+            "automation__workspace"
+        ).prefetch_related("notification_recipients")
+
     def enhance_queryset(self, queryset):
-        return queryset.prefetch_related("workflows")
+        return queryset.prefetch_related(
+            Prefetch("workflows", queryset=self._get_workflows_queryset())
+        )
 
     def enhance_and_filter_queryset(
         self,
@@ -276,9 +285,7 @@ def enhance_and_filter_queryset(
                 queryset=CoreHandler().filter_queryset(
                     user,
                     ListAutomationWorkflowsOperationType.type,
-                    AutomationWorkflow.objects.select_related(
-                        "automation__workspace"
-                    ).all(),
+                    self._get_workflows_queryset(),
                     workspace=workspace,
                 ),
             ),
diff --git a/backend/src/baserow/contrib/automation/apps.py b/backend/src/baserow/contrib/automation/apps.py
index 39223bf594..4fcfc835e7 100644
--- a/backend/src/baserow/contrib/automation/apps.py
+++ b/backend/src/baserow/contrib/automation/apps.py
@@ -93,6 +93,7 @@ def ready(self):
             action_type_registry,
         )
         from baserow.core.jobs.registries import job_type_registry
+        from baserow.core.notifications.registries import notification_type_registry
         from baserow.core.registries import (
             application_type_registry,
             object_scope_type_registry,
@@ -143,6 +144,12 @@ def ready(self):
         action_type_registry.register(ReplaceAutomationNodeActionType())
         action_type_registry.register(MoveAutomationNodeActionType())
 
+        from baserow.contrib.automation.notification_types import (
+            WorkflowDisabledNotificationType,
+        )
+
+        notification_type_registry.register(WorkflowDisabledNotificationType())
+
         action_scope_registry.register(WorkflowActionScopeType())
 
         from baserow.core.registries import permission_manager_type_registry
diff --git a/backend/src/baserow/contrib/automation/history/models.py b/backend/src/baserow/contrib/automation/history/models.py
index 0e1746eb42..ed8d8d65f7 100644
--- a/backend/src/baserow/contrib/automation/history/models.py
+++ b/backend/src/baserow/contrib/automation/history/models.py
@@ -16,7 +16,7 @@ class AutomationHistory(models.Model):
 
     class Meta:
         abstract = True
-        ordering = ("-started_on",)
+        ordering = ("-started_on", "id")
 
 
 class AutomationWorkflowHistory(AutomationHistory):
@@ -45,6 +45,18 @@ class AutomationWorkflowHistory(AutomationHistory):
         help_text="Event payload received by the workflow.",
     )
 
+    class Meta(AutomationHistory.Meta):
+        indexes = [
+            models.Index(
+                fields=["workflow", "-started_on"],
+                name="wa_hist_started_idx",
+            ),
+            models.Index(
+                fields=["workflow", "status", "-started_on"],
+                name="wa_hist_status_started_idx",
+            ),
+        ]
+
 
 class AutomationNodeHistory(AutomationHistory):
     workflow_history = models.ForeignKey(
@@ -58,7 +70,7 @@ class AutomationNodeHistory(AutomationHistory):
         related_name="node_histories",
     )
 
-    class Meta:
+    class Meta(AutomationHistory.Meta):
         indexes = [
             models.Index(fields=["workflow_history", "node"]),
         ]
@@ -71,11 +83,6 @@ class AutomationNodeResult(models.Model):
         related_name="node_results",
     )
 
-    iteration = models.PositiveIntegerField(
-        db_default=0,
-        help_text="Keeps track of the current iteration of the Iterator node.",
-    )  # TODO ZDM: Remove after next release
-
     iteration_path = models.CharField(
         db_default="",
         default="",
@@ -88,4 +95,4 @@ class AutomationNodeResult(models.Model):
     )
 
     class Meta:
-        unique_together = [["node_history", "iteration"]]
+        unique_together = [["node_history", "iteration_path"]]
diff --git a/backend/src/baserow/contrib/automation/migrations/0027_alter_automationnodehistory_options_and_more.py b/backend/src/baserow/contrib/automation/migrations/0027_alter_automationnodehistory_options_and_more.py
new file mode 100644
index 0000000000..4c30894463
--- /dev/null
+++ b/backend/src/baserow/contrib/automation/migrations/0027_alter_automationnodehistory_options_and_more.py
@@ -0,0 +1,52 @@
+# Generated by Django 5.2.13 on 2026-04-13 14:03
+
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('automation', '0026_alter_automationworkflow_unique_together'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name='automationnodehistory',
+            options={'ordering': ('-started_on', 'id')},
+        ),
+        migrations.AlterModelOptions(
+            name='automationworkflowhistory',
+            options={'ordering': ('-started_on', 'id')},
+        ),
+        migrations.AlterUniqueTogether(
+            name='automationnoderesult',
+            unique_together={('node_history', 'iteration_path')},
+        ),
+        migrations.AddIndex(
+            model_name='automationworkflowhistory',
+            index=models.Index(fields=['workflow', '-started_on'], name='wa_hist_started_idx'),
+        ),
+        migrations.AddIndex(
+            model_name='automationworkflowhistory',
+            index=models.Index(fields=['workflow', 'status', '-started_on'], name='wa_hist_status_started_idx'),
+        ),
+        migrations.RemoveField(
+            model_name='automationnoderesult',
+            name='iteration',
+        ),
+        migrations.AddField(
+            model_name='automationworkflow',
+            name='notification_recipients',
+            field=models.ManyToManyField(
+                blank=True,
+                help_text=(
+                    'The workspace users that should receive notifications related '
+                    'to this workflow.'
+                ),
+                related_name='automation_notification_workflows',
+                to=settings.AUTH_USER_MODEL,
+            ),
+        ),
+    ]
diff --git a/backend/src/baserow/contrib/automation/notification_types.py b/backend/src/baserow/contrib/automation/notification_types.py
new file mode 100644
index 0000000000..d33bb49dd8
--- /dev/null
+++ b/backend/src/baserow/contrib/automation/notification_types.py
@@ -0,0 +1,59 @@
+from dataclasses import asdict, dataclass
+from typing import List
+
+from django.utils.translation import gettext as _
+
+from baserow.contrib.automation.workflows.models import AutomationWorkflow
+from baserow.core.notifications.handler import NotificationHandler
+from baserow.core.notifications.models import NotificationRecipient
+from baserow.core.notifications.registries import NotificationType
+
+
+@dataclass
+class WorkflowDisabledNotificationData:
+    workspace_id: int
+    automation_id: int
+    workflow_id: int
+    workflow_name: str
+
+    @classmethod
+    def from_workflow(cls, workflow: AutomationWorkflow):
+        original_workflow = workflow.get_original()
+        return cls(
+            workspace_id=original_workflow.automation.workspace_id,
+            automation_id=original_workflow.automation_id,
+            workflow_id=original_workflow.id,
+            workflow_name=original_workflow.name,
+        )
+
+
+class WorkflowDisabledNotificationType(NotificationType):
+    type = "automation_workflow_disabled"
+
+    @classmethod
+    def notify_recipients(
+        cls, workflow: AutomationWorkflow
+    ) -> List[NotificationRecipient]:
+        original_workflow = workflow.get_original()
+        recipients = list(
+            original_workflow.notification_recipients.filter(
+                profile__to_be_deleted=False,
+                is_active=True,
+            ).select_related("profile")
+        )
+        if not recipients:
+            return []
+
+        return NotificationHandler.create_direct_notification_for_users(
+            notification_type=cls.type,
+            recipients=recipients,
+            data=asdict(WorkflowDisabledNotificationData.from_workflow(workflow)),
+            sender=None,
+            workspace=original_workflow.automation.workspace,
+        )
+
+    @classmethod
+    def get_notification_title(cls, notification):
+        return _("%(name)s workflow was disabled.") % {
+            "name": notification.data["workflow_name"]
+        }
diff --git a/backend/src/baserow/contrib/automation/types.py b/backend/src/baserow/contrib/automation/types.py
index 5cc234d647..91b7f9212c 100644
--- a/backend/src/baserow/contrib/automation/types.py
+++ b/backend/src/baserow/contrib/automation/types.py
@@ -12,6 +12,7 @@ class AutomationWorkflowDict(TypedDict):
     nodes: List[AutomationNodeDict]
     state: WorkflowState
     graph: dict
+    notification_recipient_emails: List[str]
 
 
 class AutomationDict(TypedDict):
diff --git a/backend/src/baserow/contrib/automation/workflows/exceptions.py b/backend/src/baserow/contrib/automation/workflows/exceptions.py
index 80167b9d52..f5709f686f 100644
--- a/backend/src/baserow/contrib/automation/workflows/exceptions.py
+++ b/backend/src/baserow/contrib/automation/workflows/exceptions.py
@@ -23,6 +23,13 @@ class AutomationWorkflowDoesNotExist(AutomationWorkflowError):
     pass
 
 
+class AutomationWorkflowNotificationRecipientsInvalid(AutomationWorkflowError):
+    """
+    Raised when a workflow notification recipient doesn't belong to the workflow
+    workspace.
+    """
+
+
 class AutomationWorkflowBeforeRunError(AutomationWorkflowError):
     pass
 
diff --git a/backend/src/baserow/contrib/automation/workflows/handler.py b/backend/src/baserow/contrib/automation/workflows/handler.py
index a3fe78d24f..b06383e4f7 100644
--- a/backend/src/baserow/contrib/automation/workflows/handler.py
+++ b/backend/src/baserow/contrib/automation/workflows/handler.py
@@ -60,6 +60,8 @@
     Progress,
     extract_allowed,
     find_unused_name,
+    set_allowed_m2m_fields,
+    split_attrs_and_m2m_fields,
 )
 
 WORKFLOW_HISTORY_RATE_LIMIT_CACHE_PREFIX = "automation_workflow_history_{}"
@@ -69,7 +71,12 @@
 
 
 class AutomationWorkflowHandler(metaclass=baserow_trace_methods(tracer)):
-    allowed_fields = ["name", "allow_test_run_until", "state"]
+    allowed_fields = [
+        "name",
+        "allow_test_run_until",
+        "state",
+        "notification_recipients",
+    ]
 
     def get_workflow(
         self,
@@ -193,7 +200,17 @@ def export_prepared_values(self, workflow: AutomationWorkflow) -> Dict[Any, Any]
         :return: A dict of prepared values.
         """
 
-        return {key: getattr(workflow, key) for key in self.allowed_fields}
+        prepared_values = {}
+        for key in self.allowed_fields:
+            if key == "notification_recipients":
+                prepared_values[key] = list(
+                    workflow.notification_recipients.order_by("id").values_list(
+                        "id", flat=True
+                    )
+                )
+            else:
+                prepared_values[key] = getattr(workflow, key)
+        return prepared_values
 
     def update_workflow(
         self, workflow: AutomationWorkflow, **kwargs
@@ -210,6 +227,9 @@ def update_workflow(
         original_workflow_values = self.export_prepared_values(workflow)
 
         allowed_values = extract_allowed(kwargs, self.allowed_fields)
+        attr_fields, m2m_fields = split_attrs_and_m2m_fields(
+            self.allowed_fields, workflow
+        )
 
         # The state is a special value that should only be set on the
         # published workflow, if available.
@@ -219,10 +239,11 @@ def update_workflow(
                 published_workflow.state = WorkflowState(state)
                 published_workflow.save(update_fields=["state"])
 
-        for key, value in allowed_values.items():
+        for key, value in extract_allowed(allowed_values, attr_fields).items():
             setattr(workflow, key, value)
 
         workflow.save()
+        set_allowed_m2m_fields(allowed_values, m2m_fields, workflow)
 
         new_workflow_values = self.export_prepared_values(workflow)
 
@@ -286,7 +307,16 @@ def duplicate_workflow(
 
         automation = workflow.automation
 
-        exported_workflow = self.export_workflow(workflow)
+        import_export_config = ImportExportConfig(
+            include_permission_data=True,
+            reduce_disk_space_usage=False,
+            exclude_sensitive_data=False,
+            is_duplicate=True,
+        )
+
+        exported_workflow = self.export_workflow(
+            workflow, import_export_config=import_export_config
+        )
 
         # Set a unique name for the workflow to import back as a new one.
         exported_workflow["name"] = self.find_unused_workflow_name(
@@ -299,13 +329,6 @@ def duplicate_workflow(
         id_mapping = defaultdict(lambda: MirrorDict())
         id_mapping["automation_workflows"] = MirrorDict()
 
-        import_export_config = ImportExportConfig(
-            include_permission_data=True,
-            reduce_disk_space_usage=False,
-            exclude_sensitive_data=False,
-            is_duplicate=True,
-        )
-
         new_workflow_clone = self.import_workflow(
             automation,
             exported_workflow,
@@ -342,14 +365,17 @@ def find_unused_workflow_name(
     def export_workflow(
         self,
         workflow: AutomationWorkflow,
+        import_export_config: Optional[ImportExportConfig] = None,
         files_zip: Optional[ExportZipFile] = None,
         storage: Optional[Storage] = None,
-        cache: Optional[Dict[str, any]] = None,
+        cache: Optional[Dict[str, Any]] = None,
     ) -> AutomationWorkflowDict:
         """
         Serializes the given workflow.
 
         :param workflow: The AutomationWorkflow instance to serialize.
+        :param import_export_config: provides configuration options for the
+            import/export process to customize how it works.
         :param files_zip: A zip file to store files in necessary.
         :param storage: Storage to use.
         :param cache: A cache to use for storing temporary data.
@@ -370,6 +396,16 @@ def export_workflow(
             nodes=serialized_nodes,
             state=workflow.state,
             graph=workflow.graph,
+            # Keep user emails only on duplication
+            notification_recipient_emails=(
+                []
+                if not getattr(import_export_config, "is_duplicate", False)
+                else list(
+                    workflow.notification_recipients.order_by("id").values_list(
+                        "email", flat=True
+                    )
+                )
+            ),
         )
 
     def _ops_count_for_import_workflow(
@@ -496,7 +532,7 @@ def import_workflow(
         files_zip: Optional[ZipFile] = None,
         storage: Optional[Storage] = None,
         progress: Optional[ChildProgressBuilder] = None,
-        cache: Optional[Dict[str, any]] = None,
+        cache: Optional[Dict[str, Any]] = None,
     ) -> AutomationWorkflow:
         """
         Creates an instance of AutomationWorkflow using the serialized version
@@ -532,7 +568,6 @@ def import_workflow_only(
         serialized_workflow: Dict[str, Any],
         id_mapping: Dict[str, Dict[int, int]],
         progress: Optional[ChildProgressBuilder] = None,
-        *args: Any,
         **kwargs: Any,
     ):
         if "automation_workflows" not in id_mapping:
@@ -545,6 +580,14 @@ def import_workflow_only(
             state=serialized_workflow["state"] or WorkflowState.DRAFT,
             graph=serialized_workflow.get("graph", {}),
         )
+        if (
+            recipient_emails := serialized_workflow.get(
+                "notification_recipient_emails", []
+            )
+        ) and (workspace := workflow_instance.get_original().automation.workspace):
+            workflow_instance.notification_recipients.set(
+                workspace.users.filter(email__in=recipient_emails)
+            )
 
         id_mapping["automation_workflows"][serialized_workflow["id"]] = (
             workflow_instance.id
@@ -657,6 +700,15 @@ def disable_workflow(self, workflow: AutomationWorkflow) -> None:
         original_workflow.state = WorkflowState.DISABLED
         original_workflow.save(update_fields=["state"])
 
+        if workflow != original_workflow:
+            from baserow.contrib.automation.notification_types import (
+                WorkflowDisabledNotificationType,
+            )
+
+            transaction.on_commit(
+                lambda: WorkflowDisabledNotificationType.notify_recipients(workflow)
+            )
+
         automation_workflow_updated.send(self, user=None, workflow=original_workflow)
 
     def set_workflow_temporary_states(self, workflow, simulate_until_node=None):
@@ -866,29 +918,57 @@ def _check_is_rate_limited(self, workflow: AutomationWorkflow) -> bool:
 
     def _check_too_many_errors(self, workflow: AutomationWorkflow) -> bool:
         """
-        Checks if the given workflow has too many consecutive errors. If so,
-        raises AutomationWorkflowTooManyErrors.
+        Checks if the given workflow has too many recent or consecutive errors.
+        If so, raises AutomationWorkflowTooManyErrors.
         """
 
-        original_workflow = workflow.get_original()
-
-        if original_workflow == workflow:
-            # We don't want to rate limit a test execution or a simulation
+        # Only check for live (published) workflows
+        if not workflow.automation.published_from_id:
             return False
 
         max_errors = settings.AUTOMATION_WORKFLOW_MAX_CONSECUTIVE_ERRORS
+        error_limits = settings.AUTOMATION_WORKFLOW_ERROR_LIMITS
 
-        statuses = (
-            self._get_histories_for_current_workflow_version(workflow)
-            .exclude(status=HistoryStatusChoices.STARTED)
-            .order_by("-started_on")
-            .values_list("status", flat=True)[: max_errors + 1]
-        )
+        now = timezone.now()
 
-        return (
-            len([s for s in statuses if s == HistoryStatusChoices.ERROR]) > max_errors
+        # Single query for all checks
+        qs = self._get_histories_for_current_workflow_version(workflow).exclude(
+            status=HistoryStatusChoices.STARTED
         )
 
+        if error_limits:
+            oldest_start_window = now - timedelta(
+                seconds=max(window_seconds for _, window_seconds in error_limits)
+            )
+            qs = qs.filter(started_on__gte=oldest_start_window)
+
+        histories = list(qs.order_by("-started_on").values_list("started_on", "status"))
+
+        # Consecutive errors check
+        latest_statuses = [status for _, status in histories[: max_errors + 1]]
+        latest_error_count = latest_statuses.count(HistoryStatusChoices.ERROR)
+        if latest_error_count > max_errors:
+            raise AutomationWorkflowTooManyErrors(
+                "The workflow was disabled due to too many consecutive errors. "
+                f"Limit exceeded: {max_errors} consecutive errors."
+            )
+
+        if not error_limits:
+            return False
+
+        # Time-window errors check
+        error_timestamps = [t for t, s in histories if s == HistoryStatusChoices.ERROR]
+        for max_errors_limit, window_seconds in error_limits:
+            start_window = now - timedelta(seconds=window_seconds)
+            if sum(t >= start_window for t in error_timestamps) > max_errors_limit:
+                raise AutomationWorkflowTooManyErrors(
+                    "The workflow was disabled due to too many recent errors. "
+                    f"Limit exceeded: {max_errors_limit} errors in "
+                    f"{window_seconds} seconds."
+                )
+
+        return False
+
     def before_run(self, workflow: AutomationWorkflow) -> None:
         """
         Runs pre-flight checks  and actions before a workflow is allowed to run.
@@ -909,10 +989,7 @@ def before_run(self, workflow: AutomationWorkflow) -> None:
         # We remove old history entries to avoid storing too many entries.
         self._clear_old_history(original_workflow)
 
-        if self._check_too_many_errors(workflow):
-            raise AutomationWorkflowTooManyErrors(
-                "The workflow was disabled due to too many consecutive errors."
-            )
+        self._check_too_many_errors(workflow)
 
         self._check_is_rate_limited(workflow)
 
diff --git a/backend/src/baserow/contrib/automation/workflows/models.py b/backend/src/baserow/contrib/automation/workflows/models.py
index 467cff1122..6cb9fafecb 100644
--- a/backend/src/baserow/contrib/automation/workflows/models.py
+++ b/backend/src/baserow/contrib/automation/workflows/models.py
@@ -1,5 +1,6 @@
 from typing import TYPE_CHECKING
 
+from django.conf import settings
 from django.db import models
 
 from baserow.contrib.automation.constants import WORKFLOW_NAME_MAX_LEN
@@ -76,6 +77,15 @@ class AutomationWorkflow(
     allow_test_run_until = models.DateTimeField(null=True, blank=True)
 
     graph = models.JSONField(default=dict, help_text="Contains the node graph.")
+    notification_recipients = models.ManyToManyField(
+        settings.AUTH_USER_MODEL,
+        blank=True,
+        related_name="automation_notification_workflows",
+        help_text=(
+            "The workspace users that should receive notifications related to "
+            "this workflow."
+        ),
+    )
 
     objects = AutomationWorkflowTrashManager()
     objects_and_trash = models.Manager()
@@ -95,6 +105,7 @@ def is_original(self) -> bool:
         """
         Whether this is an original workflow.
         """
+
         return not bool(self.automation.published_from_id)
 
     def get_original(self) -> "AutomationWorkflow":
@@ -104,10 +115,16 @@ def get_original(self) -> "AutomationWorkflow":
         :return: The original workflow that can be the current instance.
         """
 
-        if self.automation.published_from_id:
-            return self.automation.published_from
-        else:
-            return self
+        from .handler import AutomationWorkflowHandler
+
+        return local_cache.get(
+            f"automation_workflow_original_{self.id}",
+            lambda: AutomationWorkflowHandler().get_workflow(
+                self.automation.published_from_id
+            )
+            if self.automation.published_from_id
+            else self,
+        )
 
     def get_trigger(self) -> "AutomationTriggerNode":
         """
diff --git a/backend/src/baserow/contrib/automation/workflows/service.py b/backend/src/baserow/contrib/automation/workflows/service.py
index 7414283cf1..8ab5aef301 100644
--- a/backend/src/baserow/contrib/automation/workflows/service.py
+++ b/backend/src/baserow/contrib/automation/workflows/service.py
@@ -6,6 +6,9 @@
 from baserow.contrib.automation.models import Automation, AutomationWorkflow
 from baserow.contrib.automation.nodes.handler import AutomationNodeHandler
 from baserow.contrib.automation.operations import OrderAutomationWorkflowsOperationType
+from baserow.contrib.automation.workflows.exceptions import (
+    AutomationWorkflowNotificationRecipientsInvalid,
+)
 from baserow.contrib.automation.workflows.handler import AutomationWorkflowHandler
 from baserow.contrib.automation.workflows.operations import (
     CreateAutomationWorkflowOperationType,
@@ -33,6 +36,32 @@ class AutomationWorkflowService:
     def __init__(self):
         self.handler: AutomationWorkflowHandler = AutomationWorkflowHandler()
 
+    def _validate_notification_recipients(self, workspace, notification_recipient_ids):
+        if notification_recipient_ids is None:
+            return None
+
+        recipients = list(
+            workspace.users.filter(id__in=notification_recipient_ids).order_by("id")
+        )
+        if len(recipients) != len(set(notification_recipient_ids)):
+            raise AutomationWorkflowNotificationRecipientsInvalid(
+                "All notification recipients must belong to the workflow workspace."
+            )
+        return recipients
+
+    def _map_notification_recipient_ids(self, workspace, values):
+        values = values.copy()
+        notification_recipient_ids = values.pop("notification_recipient_ids", None)
+        if notification_recipient_ids is None:
+            return values
+
+        recipients = self._validate_notification_recipients(
+            workspace, notification_recipient_ids
+        )
+        if recipients is not None:
+            values["notification_recipients"] = recipients
+        return values
+
     def get_workflow(self, user: AbstractUser, workflow_id: int) -> AutomationWorkflow:
         """
         Returns an AutomationWorkflow instance by its ID.
@@ -83,6 +112,7 @@ def create_workflow(
         user: AbstractUser,
         automation_id: int,
         name: str,
+        notification_recipient_ids=None,
     ) -> AutomationWorkflow:
         """
         Returns a new instance of AutomationWorkflow.
@@ -90,6 +120,8 @@ def create_workflow(
         :param user: The user trying to create the workflow.
         :param automation_id: The automation workflow belongs to.
         :param name: The name of the workflow.
+        :param notification_recipient_ids: The ids of the user recipient of the
+          workflow notifications.
         :return: The newly created AutomationWorkflow instance.
         """
 
@@ -103,6 +135,14 @@ def create_workflow(
         )
 
         workflow = self.handler.create_workflow(automation, name)
+        recipients = self._validate_notification_recipients(
+            automation.workspace,
+            [user.id]
+            if notification_recipient_ids is None
+            else notification_recipient_ids,
+        )
+        if recipients is not None:
+            workflow.notification_recipients.set(recipients)
 
         automation_workflow_created.send(self, workflow=workflow, user=user)
 
@@ -156,6 +196,10 @@ def update_workflow(
             context=workflow,
         )
 
+        kwargs = self._map_notification_recipient_ids(
+            workflow.automation.workspace, kwargs
+        )
+
         updated_workflow = self.handler.update_workflow(workflow, **kwargs)
         automation_workflow_updated.send(
             self, user=user, workflow=updated_workflow.workflow
@@ -284,6 +328,8 @@ def publish(
 
         automation_workflow_published.send(self, user=user, workflow=published_workflow)
 
+        return published_workflow
+
     def toggle_test_run(
         self,
         user: AbstractUser,
diff --git a/backend/tests/baserow/contrib/automation/api/test_automation_application_views.py b/backend/tests/baserow/contrib/automation/api/test_automation_application_views.py
index db6e4b44b8..e40aa5d8a6 100644
--- a/backend/tests/baserow/contrib/automation/api/test_automation_application_views.py
+++ b/backend/tests/baserow/contrib/automation/api/test_automation_application_views.py
@@ -51,6 +51,7 @@ def test_get_automation_application(api_client, data_fixture):
                 "simulate_until_node_id": None,
                 "state": "draft",
                 "published_on": None,
+                "notification_recipient_ids": [],
                 "graph": {"0": trigger.id, str(trigger.id): {}},
             }
         ],
@@ -105,6 +106,7 @@ def test_list_automation_applications(api_client, data_fixture):
                     "simulate_until_node_id": None,
                     "state": "draft",
                     "published_on": None,
+                    "notification_recipient_ids": [],
                     "graph": {"0": trigger.id, str(trigger.id): {}},
                 }
             ],
diff --git a/backend/tests/baserow/contrib/automation/api/test_automation_serializer.py b/backend/tests/baserow/contrib/automation/api/test_automation_serializer.py
index d968b57f46..3181710438 100644
--- a/backend/tests/baserow/contrib/automation/api/test_automation_serializer.py
+++ b/backend/tests/baserow/contrib/automation/api/test_automation_serializer.py
@@ -57,5 +57,6 @@ def test_serializer_get_workflows(automation_fixture):
             "simulate_until_node_id": None,
             "published_on": None,
             "graph": {"0": trigger.id, str(trigger.id): {}},
+            "notification_recipient_ids": [],
         }
     ]
diff --git a/backend/tests/baserow/contrib/automation/api/workflows/test_automation_workflow_serializer.py b/backend/tests/baserow/contrib/automation/api/workflows/test_automation_workflow_serializer.py
index 25f4479fd5..1f5793aea1 100644
--- a/backend/tests/baserow/contrib/automation/api/workflows/test_automation_workflow_serializer.py
+++ b/backend/tests/baserow/contrib/automation/api/workflows/test_automation_workflow_serializer.py
@@ -36,6 +36,7 @@ def test_automation_workflow_serializer_fields(workflow_fixture):
         "graph",
         "id",
         "name",
+        "notification_recipient_ids",
         "order",
         "published_on",
         "simulate_until_node_id",
@@ -58,4 +59,7 @@ def test_update_automation_workflow_serializer_fields(workflow_fixture):
 
     serializer = UpdateAutomationWorkflowSerializer(instance=workflow)
 
-    assert sorted(serializer.data.keys()) == ["name", "state"]
+    assert sorted(serializer.data.keys()) == [
+        "name",
+        "state",
+    ]
diff --git a/backend/tests/baserow/contrib/automation/api/workflows/test_workflow_views.py b/backend/tests/baserow/contrib/automation/api/workflows/test_workflow_views.py
index aec2a979d9..2f5ef19d68 100644
--- a/backend/tests/baserow/contrib/automation/api/workflows/test_workflow_views.py
+++ b/backend/tests/baserow/contrib/automation/api/workflows/test_workflow_views.py
@@ -51,6 +51,7 @@ def test_create_workflow(api_client, data_fixture):
         "automation_id": AnyInt(),
         "id": AnyInt(),
         "name": name,
+        "notification_recipient_ids": [user.id],
         "order": AnyInt(),
         "state": "draft",
         "published_on": None,
@@ -143,6 +144,7 @@ def test_read_workflow(api_client, data_fixture):
         "state": "draft",
         "published_on": None,
         "graph": {"0": trigger.id, str(trigger.id): {}},
+        "notification_recipient_ids": [],
     }
 
 
@@ -160,6 +162,47 @@ def test_update_workflow(api_client, data_fixture):
     assert response.json()["name"] == "test-updated"
 
 
+@pytest.mark.django_db
+def test_update_workflow_notification_recipients(api_client, data_fixture):
+    user, token = data_fixture.create_user_and_token()
+    workflow = data_fixture.create_automation_workflow(user, name="test")
+    recipient = data_fixture.create_user(workspace=workflow.automation.workspace)
+
+    url = reverse(API_URL_WORKFLOW_ITEM, kwargs={"workflow_id": workflow.id})
+    response = api_client.patch(
+        url,
+        {"notification_recipient_ids": [recipient.id]},
+        format="json",
+        HTTP_AUTHORIZATION=f"JWT {token}",
+    )
+
+    assert response.status_code == HTTP_200_OK
+    assert response.json()["notification_recipient_ids"] == [recipient.id]
+
+
+@pytest.mark.django_db
+def test_update_workflow_rejects_non_workspace_notification_recipient(
+    api_client, data_fixture
+):
+    user, token = data_fixture.create_user_and_token()
+    workflow = data_fixture.create_automation_workflow(user, name="test")
+    recipient = data_fixture.create_user()
+
+    url = reverse(API_URL_WORKFLOW_ITEM, kwargs={"workflow_id": workflow.id})
+    response = api_client.patch(
+        url,
+        {"notification_recipient_ids": [recipient.id]},
+        format="json",
+        HTTP_AUTHORIZATION=f"JWT {token}",
+    )
+
+    assert response.status_code == HTTP_400_BAD_REQUEST
+    assert response.json() == {
+        "error": "ERROR_AUTOMATION_WORKFLOW_NOTIFICATION_RECIPIENTS_INVALID",
+        "detail": "All notification recipients must belong to the workflow workspace.",
+    }
+
+
 @pytest.mark.django_db
 def test_update_workflow_does_not_exist(api_client, data_fixture):
     _, token = data_fixture.create_user_and_token()
@@ -365,6 +408,7 @@ def test_duplicate_workflow(api_client, data_fixture):
             "automation_id": automation.id,
             "id": workflow.id,
             "name": "test",
+            "notification_recipient_ids": [],
             "order": AnyInt(),
             "state": "draft",
             "published_on": None,
diff --git a/backend/tests/baserow/contrib/automation/nodes/test_node_dispatch_async.py b/backend/tests/baserow/contrib/automation/nodes/test_node_dispatch_async.py
index a7944cf8c7..5f5806f3ee 100644
--- a/backend/tests/baserow/contrib/automation/nodes/test_node_dispatch_async.py
+++ b/backend/tests/baserow/contrib/automation/nodes/test_node_dispatch_async.py
@@ -292,7 +292,7 @@ def test_dispatch_node_dispatches_trigger(data_fixture):
     assert node_history.status == HistoryStatusChoices.SUCCESS
 
     node_result = AutomationNodeResult.objects.get(node_history=node_history)
-    assert node_result.iteration == 0
+    assert node_result.iteration_path == ""
     assert node_result.result == workflow_history.event_payload
 
     assert_dispatches_next_node(result, (action_node, workflow_history, None))
@@ -344,7 +344,7 @@ def test_dispatch_node_dispatches_action_create_row(data_fixture):
     assert node_history.status == HistoryStatusChoices.SUCCESS
 
     node_result = AutomationNodeResult.objects.get(node_history=node_history)
-    assert node_result.iteration == 0
+    assert node_result.iteration_path == ""
     assert node_result.result == {
         action_table_field.name: "Apple",
         "id": AnyInt(),
@@ -611,7 +611,7 @@ def test_dispatch_node_dispatches_action_simulation(
     assert node_history.status == HistoryStatusChoices.SUCCESS
 
     node_result = AutomationNodeResult.objects.get(node_history=node_history)
-    assert node_result.iteration == 0
+    assert node_result.iteration_path == ""
     assert node_result.result == {
         "results": [
             {
@@ -923,12 +923,19 @@ def test_dispatch_node_dispatches_test_run(
     assert workflow_history.status == HistoryStatusChoices.STARTED
 
     # Make sure all nodes have a history and node result
-    for node in [
-        trigger_node,
-        iterator_node,
-        iterator_child_1_node,
-        iterator_child_2_node,
-    ]:
+    for node in [trigger_node, iterator_node]:
+        for node_history in AutomationNodeHistory.objects.filter(
+            workflow_history=workflow_history,
+            node=node,
+        ):
+            assert node_history.message == ""
+            assert node_history.status == HistoryStatusChoices.SUCCESS
+
+            node_result = AutomationNodeResult.objects.get(node_history=node_history)
+            assert node_result.iteration_path == ""
+            assert len(node_result.result) > 0
+
+    for node in [iterator_child_1_node, iterator_child_2_node]:
         for index, node_history in enumerate(
             AutomationNodeHistory.objects.filter(
                 workflow_history=workflow_history,
@@ -939,7 +946,7 @@ def test_dispatch_node_dispatches_test_run(
             assert node_history.status == HistoryStatusChoices.SUCCESS
 
             node_result = AutomationNodeResult.objects.get(node_history=node_history)
-            assert node_result.iteration == index
+            assert node_result.iteration_path == str(index)
             assert len(node_result.result) > 0
 
     # Ensure workflow history is exists for test runs
@@ -1010,7 +1017,7 @@ def test_dispatch_node_dispatches_action_update_row(data_fixture):
     assert node_history.status == HistoryStatusChoices.SUCCESS
 
     node_result = AutomationNodeResult.objects.get(node_history=node_history)
-    assert node_result.iteration == 0
+    assert node_result.iteration_path == ""
     assert node_result.result == {
         action_table_field.name: "Apple",
         "id": AnyInt(),
@@ -1057,7 +1064,7 @@ def test_dispatch_node_dispatches_action_delete_row(data_fixture):
     assert node_history.status == HistoryStatusChoices.SUCCESS
 
     node_result = AutomationNodeResult.objects.get(node_history=node_history)
-    assert node_result.iteration == 0
+    assert node_result.iteration_path == ""
     assert node_result.result == {}
 
 
@@ -1142,7 +1149,7 @@ def test_dispatch_node_dispatches_action_router(data_fixture):
     assert node_history.status == HistoryStatusChoices.SUCCESS
 
     node_result = AutomationNodeResult.objects.get(node_history=node_history)
-    assert node_result.iteration == 0
+    assert node_result.iteration_path == ""
     assert node_result.result == {
         action_table_field.name: "Cherry",
         "id": AnyInt(),
diff --git a/backend/tests/baserow/contrib/automation/test_automation_application_types.py b/backend/tests/baserow/contrib/automation/test_automation_application_types.py
index af6f1242ac..c9b7674d62 100644
--- a/backend/tests/baserow/contrib/automation/test_automation_application_types.py
+++ b/backend/tests/baserow/contrib/automation/test_automation_application_types.py
@@ -105,6 +105,7 @@ def test_automation_export_serialized(data_fixture):
                 "name": workflow.name,
                 "order": workflow.order,
                 "state": workflow.state,
+                "notification_recipient_emails": [],
                 "nodes": [
                     {
                         "id": trigger.id,
@@ -255,21 +256,31 @@ def test_fetch_workflows_to_serialize_with_user(data_fixture):
 @pytest.mark.django_db
 def test_enhance_queryset(data_fixture, django_assert_num_queries):
     user = data_fixture.create_user()
+    recipient = data_fixture.create_user()
 
     # Create two automations with 2 workflows each
     workflow_1 = data_fixture.create_automation_workflow(user=user)
-    data_fixture.create_automation_workflow(automation=workflow_1.automation, user=user)
+    workflow_1.notification_recipients.add(recipient)
+    workflow_2 = data_fixture.create_automation_workflow(
+        automation=workflow_1.automation, user=user
+    )
+    workflow_2.notification_recipients.add(recipient)
 
     workflow_3 = data_fixture.create_automation_workflow(user=user)
-    data_fixture.create_automation_workflow(automation=workflow_3.automation, user=user)
+    workflow_3.notification_recipients.add(recipient)
+    workflow_4 = data_fixture.create_automation_workflow(
+        automation=workflow_3.automation, user=user
+    )
+    workflow_4.notification_recipients.add(recipient)
 
     # query 1: fetch all automations
     # query 2: fetch all workflows for automation 1
     # query 3: fetch all workflows for automation 2
-    expected_queries = 3
+    # query 4-7: fetch notification recipients for each workflow
+    expected_queries = 7
     with django_assert_num_queries(expected_queries):
         [
-            workflow.name
+            sorted(recipient.id for recipient in workflow.notification_recipients.all())
             for automation in Automation.objects.all()
             for workflow in automation.workflows.all()
         ]
@@ -277,11 +288,12 @@ def test_enhance_queryset(data_fixture, django_assert_num_queries):
     queryset = AutomationApplicationType().enhance_queryset(Automation.objects.all())
 
     # query 1: fetch all automations
-    # query 2: fetch all workflows for automation 2
-    expected_queries = 2
+    # query 2: fetch all workflows
+    # query 3: fetch all notification recipients for prefetched workflows
+    expected_queries = 3
     with django_assert_num_queries(expected_queries):
         [
-            workflow.name
+            sorted(recipient.id for recipient in workflow.notification_recipients.all())
             for automation in queryset
             for workflow in automation.workflows.all()
         ]
diff --git a/backend/tests/baserow/contrib/automation/workflows/test_workflow_handler.py b/backend/tests/baserow/contrib/automation/workflows/test_workflow_handler.py
index 3b2b8d9b0c..5c9756bb5e 100644
--- a/backend/tests/baserow/contrib/automation/workflows/test_workflow_handler.py
+++ b/backend/tests/baserow/contrib/automation/workflows/test_workflow_handler.py
@@ -28,6 +28,9 @@
     AutomationWorkflowTooManyErrors,
 )
 from baserow.contrib.automation.workflows.handler import AutomationWorkflowHandler
+from baserow.core.cache import local_cache
+from baserow.core.notifications.models import Notification, NotificationRecipient
+from baserow.core.registries import ImportExportConfig
 from baserow.core.trash.handler import TrashHandler
 from tests.baserow.contrib.automation.history.utils import assert_history
 
@@ -124,6 +127,83 @@ def test_create_workflow_integrity_error(data_fixture):
         assert str(exc_info.value) == "unexpected integrity error"
 
 
+@pytest.mark.django_db
+def test_export_workflow_excludes_notification_recipients_when_not_duplicating(
+    data_fixture,
+):
+    user = data_fixture.create_user()
+    automation = data_fixture.create_automation_application(user=user)
+    recipient = data_fixture.create_user(workspace=automation.workspace)
+    workflow = data_fixture.create_automation_workflow(automation=automation)
+    workflow.notification_recipients.add(recipient)
+
+    exported_workflow = AutomationWorkflowHandler().export_workflow(
+        workflow,
+        import_export_config=ImportExportConfig(
+            include_permission_data=True,
+            is_duplicate=False,
+        ),
+    )
+
+    assert exported_workflow["notification_recipient_emails"] == []
+
+
+@pytest.mark.django_db
+def test_exported_workflow_import_maps_notification_recipients_to_target_workspace(
+    data_fixture,
+):
+    source_workspace = data_fixture.create_workspace()
+    target_workspace = data_fixture.create_workspace()
+
+    shared_recipient = data_fixture.create_user(
+        email="shared-recipient@example.com", workspace=source_workspace
+    )
+    source_only_recipient = data_fixture.create_user(
+        email="source-only-recipient@example.com", workspace=source_workspace
+    )
+    target_only_user = data_fixture.create_user(
+        email="target-only-user@example.com", workspace=target_workspace
+    )
+    data_fixture.create_user_workspace(
+        workspace=target_workspace, user=shared_recipient
+    )
+
+    source_automation = data_fixture.create_automation_application(
+        workspace=source_workspace
+    )
+    source_workflow = data_fixture.create_automation_workflow(
+        automation=source_automation,
+        create_trigger=False,
+    )
+    source_workflow.notification_recipients.add(shared_recipient, source_only_recipient)
+
+    exported_workflow = AutomationWorkflowHandler().export_workflow(
+        source_workflow,
+        import_export_config=ImportExportConfig(
+            include_permission_data=True,
+            is_duplicate=True,
+        ),
+    )
+
+    assert exported_workflow["notification_recipient_emails"] == [
+        shared_recipient.email,
+        source_only_recipient.email,
+    ]
+
+    target_automation = data_fixture.create_automation_application(
+        workspace=target_workspace
+    )
+
+    imported_workflow = AutomationWorkflowHandler().import_workflow(
+        target_automation,
+        exported_workflow,
+        {},
+    )
+
+    assert list(imported_workflow.notification_recipients.all()) == [shared_recipient]
+    assert target_only_user not in imported_workflow.notification_recipients.all()
+
+
 @patch(f"{TRASH_TYPES_PATH}.automation_workflow_deleted")
 @pytest.mark.django_db
 def test_delete_workflow(workflow_deleted_mock, data_fixture):
@@ -323,6 +403,7 @@ def test_export_prepared_values(data_fixture):
         "name": "test",
         "allow_test_run_until": None,
         "state": WorkflowState.DRAFT,
+        "notification_recipients": [],
     }
 
 
@@ -454,12 +535,14 @@ def test_update_workflow_correctly_pauses_published_workflow(data_fixture):
         "name": "foo",
         "allow_test_run_until": None,
         "state": WorkflowState.DRAFT,
+        "notification_recipients": [],
     }
     assert updated.new_values == {
         "name": "foo",
         "allow_test_run_until": None,
         # The original workflow should indeed be unaffected
         "state": WorkflowState.DRAFT,
+        "notification_recipients": [],
     }
 
     published_workflow.refresh_from_db()
@@ -480,6 +563,29 @@ def test_get_original_workflow_returns_original_workflow(data_fixture):
     assert workflow == original_workflow
 
 
+@pytest.mark.django_db
+def test_get_original_workflow_uses_local_cache(
+    data_fixture, django_assert_num_queries
+):
+    original_workflow = data_fixture.create_automation_workflow()
+    published_workflow = data_fixture.create_automation_workflow(
+        state=WorkflowState.LIVE
+    )
+    published_workflow.automation.published_from = original_workflow
+    published_workflow.automation.save()
+
+    with local_cache.context():
+        cached_original_workflow = published_workflow.get_original()
+        reloaded_published_workflow = AutomationWorkflow.objects.select_related(
+            "automation"
+        ).get(id=published_workflow.id)
+
+        with django_assert_num_queries(0):
+            assert (
+                reloaded_published_workflow.get_original() is cached_original_workflow
+            )
+
+
 @pytest.mark.django_db
 def test_trashing_workflow_deletes_published_workflow(data_fixture):
     user = data_fixture.create_user()
@@ -778,6 +884,53 @@ def test_disable_workflow_disables_published_workflow(data_fixture):
     assert original_workflow.state == WorkflowState.DISABLED
 
 
+@pytest.mark.django_db(transaction=True)
+def test_disable_workflow_notifies_selected_recipients(data_fixture):
+    workspace = data_fixture.create_workspace()
+    original_workflow = data_fixture.create_automation_workflow(
+        automation=data_fixture.create_automation_application(workspace=workspace)
+    )
+    recipient_1 = data_fixture.create_user(workspace=workspace)
+    recipient_2 = data_fixture.create_user(workspace=workspace)
+    published_workflow = data_fixture.create_automation_workflow(
+        automation=data_fixture.create_automation_application(workspace=workspace),
+        state=WorkflowState.LIVE,
+    )
+    published_workflow.automation.published_from = original_workflow
+    published_workflow.automation.save()
+    original_workflow.notification_recipients.add(recipient_1, recipient_2)
+
+    AutomationWorkflowHandler().disable_workflow(published_workflow)
+
+    notification = Notification.objects.get(type="automation_workflow_disabled")
+    assert notification.workspace == workspace
+    assert notification.data == {
+        "workspace_id": workspace.id,
+        "automation_id": original_workflow.automation_id,
+        "workflow_id": original_workflow.id,
+        "workflow_name": original_workflow.name,
+    }
+    assert set(
+        NotificationRecipient.objects.filter(notification=notification).values_list(
+            "recipient_id", flat=True
+        )
+    ) == {recipient_1.id, recipient_2.id}
+
+
+@pytest.mark.django_db(transaction=True)
+def test_disable_workflow_with_no_selected_recipients_does_not_notify(data_fixture):
+    original_workflow = data_fixture.create_automation_workflow()
+    published_workflow = data_fixture.create_automation_workflow(
+        state=WorkflowState.LIVE
+    )
+    published_workflow.automation.published_from = original_workflow
+    published_workflow.automation.save()
+
+    AutomationWorkflowHandler().disable_workflow(published_workflow)
+
+    assert Notification.objects.filter(type="automation_workflow_disabled").count() == 0
+
+
 @override_settings(AUTOMATION_WORKFLOW_MAX_CONSECUTIVE_ERRORS=5)
 @pytest.mark.django_db
 def test_check_too_many_errors_raises_if_above_limit(data_fixture):
@@ -798,14 +951,18 @@ def test_check_too_many_errors_raises_if_above_limit(data_fixture):
         AutomationWorkflowHandler()._check_too_many_errors(published_workflow) is False
     )
 
-    # This 6th error should cause True to be returned
+    # This 6th error should exceed the configured consecutive error limit.
     data_fixture.create_automation_workflow_history(
         workflow=original_workflow,
         status=HistoryStatusChoices.ERROR,
     )
 
-    assert (
-        AutomationWorkflowHandler()._check_too_many_errors(published_workflow) is True
+    with pytest.raises(AutomationWorkflowTooManyErrors) as exc:
+        AutomationWorkflowHandler()._check_too_many_errors(published_workflow)
+
+    assert str(exc.value) == (
+        "The workflow was disabled due to too many consecutive errors. "
+        "Limit exceeded: 5 consecutive errors."
     )
 
 
@@ -876,6 +1033,146 @@ def test_check_too_many_errors_returns_none_if_below_limit(data_fixture):
     AutomationWorkflowHandler()._check_too_many_errors(original_workflow)
 
 
+@override_settings(
+    AUTOMATION_WORKFLOW_MAX_CONSECUTIVE_ERRORS=10,
+    AUTOMATION_WORKFLOW_ERROR_LIMITS=((5, 30),),
+)
+@pytest.mark.django_db
+def test_check_too_many_errors_raises_if_above_error_window_limit(data_fixture):
+    original_workflow = data_fixture.create_automation_workflow()
+
+    with freeze_time("2026-03-10 09:59:00"):
+        published_workflow = data_fixture.create_automation_workflow(
+            state=WorkflowState.LIVE
+        )
+        published_workflow.automation.published_from = original_workflow
+        published_workflow.automation.save()
+
+    with freeze_time("2026-03-10 10:00:00"):
+        for _ in range(5):
+            data_fixture.create_automation_workflow_history(
+                workflow=original_workflow,
+                status=HistoryStatusChoices.ERROR,
+            )
+
+        assert (
+            AutomationWorkflowHandler()._check_too_many_errors(published_workflow)
+            is False
+        )
+
+        data_fixture.create_automation_workflow_history(
+            workflow=original_workflow,
+            status=HistoryStatusChoices.ERROR,
+        )
+
+        with pytest.raises(AutomationWorkflowTooManyErrors) as exc:
+            AutomationWorkflowHandler()._check_too_many_errors(published_workflow)
+
+        assert str(exc.value) == (
+            "The workflow was disabled due to too many recent errors. "
+            "Limit exceeded: 5 errors in 30 seconds."
+        )
+
+
+@override_settings(
+    AUTOMATION_WORKFLOW_MAX_CONSECUTIVE_ERRORS=10,
+    AUTOMATION_WORKFLOW_ERROR_LIMITS=((2, 5), (4, 60)),
+)
+@pytest.mark.django_db
+def test_check_too_many_errors_uses_multiple_error_windows(data_fixture):
+    original_workflow = data_fixture.create_automation_workflow()
+    with freeze_time("2026-03-10 09:59:00"):
+        published_workflow = data_fixture.create_automation_workflow(
+            state=WorkflowState.LIVE
+        )
+        published_workflow.automation.published_from = original_workflow
+        published_workflow.automation.save()
+
+    with freeze_time("2026-03-10 10:00:00"):
+        for _ in range(3):
+            data_fixture.create_automation_workflow_history(
+                workflow=original_workflow,
+                status=HistoryStatusChoices.ERROR,
+            )
+
+    with freeze_time("2026-03-10 10:00:10"):
+        assert (
+            AutomationWorkflowHandler()._check_too_many_errors(published_workflow)
+            is False
+        )
+
+    with freeze_time("2026-03-10 10:00:30"):
+        for _ in range(2):
+            data_fixture.create_automation_workflow_history(
+                workflow=original_workflow,
+                status=HistoryStatusChoices.ERROR,
+            )
+
+        with pytest.raises(AutomationWorkflowTooManyErrors) as exc:
+            AutomationWorkflowHandler()._check_too_many_errors(published_workflow)
+
+        assert str(exc.value) == (
+            "The workflow was disabled due to too many recent errors. "
+            "Limit exceeded: 4 errors in 60 seconds."
+        )
+
+
+@override_settings(
+    AUTOMATION_WORKFLOW_MAX_CONSECUTIVE_ERRORS=10,
+    AUTOMATION_WORKFLOW_ERROR_LIMITS=((2, 5), (4, 60), (10, 3600)),
+)
+@pytest.mark.django_db
+def test_check_too_many_errors_uses_a_single_query_for_multiple_windows(data_fixture):
+    original_workflow = data_fixture.create_automation_workflow()
+    with freeze_time("2026-03-10 09:59:00"):
+        published_workflow = data_fixture.create_automation_workflow(
+            state=WorkflowState.LIVE
+        )
+        published_workflow.automation.published_from = original_workflow
+        published_workflow.automation.save()
+
+    with freeze_time("2026-03-10 10:00:00"):
+        for _ in range(5):
+            data_fixture.create_automation_workflow_history(
+                workflow=original_workflow,
+                status=HistoryStatusChoices.ERROR,
+            )
+
+        with CaptureQueriesContext(connection) as queries:
+            with pytest.raises(AutomationWorkflowTooManyErrors) as exc:
+                AutomationWorkflowHandler()._check_too_many_errors(published_workflow)
+
+        assert str(exc.value) == (
+            "The workflow was disabled due to too many recent errors. "
+            "Limit exceeded: 2 errors in 5 seconds."
+        )
+
+    assert len(queries) == 2
+
+
+@override_settings(
+    AUTOMATION_WORKFLOW_MAX_CONSECUTIVE_ERRORS=10,
+    AUTOMATION_WORKFLOW_ERROR_LIMITS=((2, 30),),
+)
+@pytest.mark.django_db
+def test_check_too_many_errors_ignores_errors_before_latest_publish_for_windows(
+    data_fixture,
+):
+    original_workflow = data_fixture.create_automation_workflow()
+    handler = AutomationWorkflowHandler()
+
+    with freeze_time("2026-03-10 10:00:00"):
+        for _ in range(3):
+            data_fixture.create_automation_workflow_history(
+                workflow=original_workflow,
+                status=HistoryStatusChoices.ERROR,
+            )
+
+    with freeze_time("2026-03-10 12:00:00"):
+        published_workflow = handler.publish(original_workflow)
+        assert handler._check_too_many_errors(published_workflow) is False
+
+
 @patch(f"{WORKFLOWS_MODULE}.handler.automation_workflow_updated")
 @patch(f"{WORKFLOWS_MODULE}.handler.AutomationWorkflowHandler.async_start_workflow")
 @pytest.mark.django_db
@@ -1242,7 +1539,8 @@ def test_async_start_workflow_rate_limited_runs_eventually_disable_workflow(
 
     assert histories[5].status == HistoryStatusChoices.DISABLED
     assert histories[5].message == (
-        "The workflow was disabled due to too many consecutive errors."
+        "The workflow was disabled due to too many consecutive errors. "
+        "Limit exceeded: 2 consecutive errors."
     )
 
     original_workflow.refresh_from_db()
diff --git a/backend/tests/baserow/contrib/automation/workflows/test_workflow_service.py b/backend/tests/baserow/contrib/automation/workflows/test_workflow_service.py
index 60b6832eef..ebf08ee911 100644
--- a/backend/tests/baserow/contrib/automation/workflows/test_workflow_service.py
+++ b/backend/tests/baserow/contrib/automation/workflows/test_workflow_service.py
@@ -2,10 +2,11 @@
 
 import pytest
 
-from baserow.contrib.automation.models import Automation, AutomationWorkflow
+from baserow.contrib.automation.models import AutomationWorkflow
 from baserow.contrib.automation.workflows.constants import WorkflowState
 from baserow.contrib.automation.workflows.exceptions import (
     AutomationWorkflowDoesNotExist,
+    AutomationWorkflowNotificationRecipientsInvalid,
     AutomationWorkflowNotInAutomation,
 )
 from baserow.contrib.automation.workflows.service import AutomationWorkflowService
@@ -76,6 +77,26 @@ def test_create_workflow(data_fixture):
     workflow = AutomationWorkflowService().create_workflow(user, automation.id, "foo")
 
     assert workflow.automation_workflow_nodes.count() == 0
+    assert list(workflow.notification_recipients.all()) == [user]
+
+
+@pytest.mark.django_db
+def test_create_workflow_rejects_non_workspace_notification_recipients(data_fixture):
+    user = data_fixture.create_user()
+    automation = data_fixture.create_automation_application(user=user)
+    recipient = data_fixture.create_user()
+
+    with pytest.raises(AutomationWorkflowNotificationRecipientsInvalid) as exc:
+        AutomationWorkflowService().create_workflow(
+            user,
+            automation.id,
+            "foo",
+            notification_recipient_ids=[recipient.id],
+        )
+
+    assert str(exc.value) == (
+        "All notification recipients must belong to the workflow workspace."
+    )
 
 
 @patch(f"{SERVICES_PATH}.automation_workflow_deleted")
@@ -150,6 +171,37 @@ def test_update_workflow_ignores_invalid_values(data_fixture):
     assert hasattr(updated_workflow, "foo") is False
 
 
+@pytest.mark.django_db
+def test_update_workflow_updates_notification_recipients(data_fixture):
+    user = data_fixture.create_user()
+    automation = data_fixture.create_automation_application(user=user)
+    workflow = data_fixture.create_automation_workflow(automation=automation)
+    recipient = data_fixture.create_user(workspace=automation.workspace)
+
+    updated_workflow = AutomationWorkflowService().update_workflow(
+        user, workflow.id, notification_recipient_ids=[recipient.id]
+    )
+
+    assert list(updated_workflow.workflow.notification_recipients.all()) == [recipient]
+
+
+@pytest.mark.django_db
+def test_update_workflow_rejects_non_workspace_notification_recipients(data_fixture):
+    user = data_fixture.create_user()
+    automation = data_fixture.create_automation_application(user=user)
+    workflow = data_fixture.create_automation_workflow(automation=automation)
+    recipient = data_fixture.create_user()
+
+    with pytest.raises(AutomationWorkflowNotificationRecipientsInvalid) as exc:
+        AutomationWorkflowService().update_workflow(
+            user, workflow.id, notification_recipient_ids=[recipient.id]
+        )
+
+    assert str(exc.value) == (
+        "All notification recipients must belong to the workflow workspace."
+    )
+
+
 @patch(f"{SERVICES_PATH}.automation_workflows_reordered")
 @pytest.mark.django_db
 def test_workflows_reordered_signal_sent(workflows_reordered_mock, data_fixture):
@@ -207,6 +259,19 @@ def test_duplicate_workflow(data_fixture):
     assert workflow_clone.name != workflow.name
 
 
+@pytest.mark.django_db
+def test_duplicate_workflow_preserves_notification_recipients(data_fixture):
+    user = data_fixture.create_user()
+    automation = data_fixture.create_automation_application(user=user)
+    recipient = data_fixture.create_user(workspace=automation.workspace)
+    workflow = data_fixture.create_automation_workflow(automation=automation)
+    workflow.notification_recipients.add(recipient)
+
+    workflow_clone = AutomationWorkflowService().duplicate_workflow(user, workflow)
+
+    assert list(workflow_clone.notification_recipients.all()) == [recipient]
+
+
 @pytest.mark.django_db
 def test_duplicate_workflow_user_not_in_workspace(data_fixture):
     user = data_fixture.create_user()
@@ -256,18 +321,32 @@ def test_publish_workflow(mock_signal, data_fixture):
     workflow = data_fixture.create_automation_workflow(automation=automation)
 
     service = AutomationWorkflowService()
-    service.publish(user, workflow, Progress(0))
+    published_workflow = service.publish(user, workflow, Progress(0))
 
     workflow.refresh_from_db()
     assert workflow.state == WorkflowState.DRAFT
 
-    published_automation = Automation.objects.get(published_from=workflow)
+    published_automation = published_workflow.automation
     assert published_automation.automation.workspace is None
     assert published_automation.workflows.count() == 1
 
-    published_workflow = published_automation.workflows.first()
     assert published_workflow.state == WorkflowState.LIVE
 
     mock_signal.send.assert_called_once_with(
         service, user=user, workflow=published_workflow
     )
+
+
+@pytest.mark.django_db
+def test_publish_workflow_removes_notification_recipients(data_fixture):
+    user = data_fixture.create_user()
+    automation = data_fixture.create_automation_application(user=user)
+    workflow = data_fixture.create_automation_workflow(automation=automation)
+    recipient = data_fixture.create_user(workspace=automation.workspace)
+    workflow.notification_recipients.add(recipient)
+
+    published_workflow = AutomationWorkflowService().publish(
+        user, workflow, Progress(0)
+    )
+
+    assert list(published_workflow.notification_recipients.all()) == []
diff --git a/changelog/entries/unreleased/feature/5186_send_notification_when_a_workflow_is_disabled.json b/changelog/entries/unreleased/feature/5186_send_notification_when_a_workflow_is_disabled.json
new file mode 100644
index 0000000000..3385819fad
--- /dev/null
+++ b/changelog/entries/unreleased/feature/5186_send_notification_when_a_workflow_is_disabled.json
@@ -0,0 +1,9 @@
+{
+    "type": "feature",
+    "message": "Send notification when a workflow is disabled",
+    "issue_origin": "github",
+    "issue_number": 5186,
+    "domain": "automation",
+    "bullet_points": [],
+    "created_at": "2026-04-14"
+}
\ No newline at end of file
diff --git a/docker-compose.no-caddy.yml b/docker-compose.no-caddy.yml
index c8e3a5c488..879391adaf 100644
--- a/docker-compose.no-caddy.yml
+++ b/docker-compose.no-caddy.yml
@@ -90,6 +90,7 @@ x-backend-variables:
   BASEROW_AUTOMATION_WORKFLOW_RATE_LIMITS:
   BASEROW_AUTOMATION_WORKFLOW_RATE_LIMIT_CACHE_EXPIRY_SECONDS:
   BASEROW_AUTOMATION_WORKFLOW_HISTORY_RATE_LIMIT_CACHE_EXPIRY_SECONDS:
+  BASEROW_AUTOMATION_WORKFLOW_ERROR_LIMITS:
   BASEROW_AUTOMATION_WORKFLOW_MAX_CONSECUTIVE_ERRORS:
   BASEROW_AUTOMATION_WORKFLOW_TIMEOUT_HOURS:
   BASEROW_AUTOMATION_WORKFLOW_HISTORY_MAX_DAYS:
@@ -255,6 +256,7 @@ services:
       BASEROW_AUTOMATION_WORKFLOW_RATE_LIMITS:
       BASEROW_AUTOMATION_WORKFLOW_RATE_LIMIT_CACHE_EXPIRY_SECONDS:
       BASEROW_AUTOMATION_WORKFLOW_HISTORY_RATE_LIMIT_CACHE_EXPIRY_SECONDS:
+      BASEROW_AUTOMATION_WORKFLOW_ERROR_LIMITS:
       BASEROW_AUTOMATION_WORKFLOW_MAX_CONSECUTIVE_ERRORS:
       BASEROW_AUTOMATION_WORKFLOW_TIMEOUT_HOURS:
       BASEROW_AUTOMATION_WORKFLOW_HISTORY_MAX_DAYS:
diff --git a/docker-compose.yml b/docker-compose.yml
index 65712e340d..b2d761014a 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -104,6 +104,7 @@ x-backend-variables:
   BASEROW_AUTOMATION_WORKFLOW_RATE_LIMITS:
   BASEROW_AUTOMATION_WORKFLOW_RATE_LIMIT_CACHE_EXPIRY_SECONDS:
   BASEROW_AUTOMATION_WORKFLOW_HISTORY_RATE_LIMIT_CACHE_EXPIRY_SECONDS:
+  BASEROW_AUTOMATION_WORKFLOW_ERROR_LIMITS:
   BASEROW_AUTOMATION_WORKFLOW_MAX_CONSECUTIVE_ERRORS:
   BASEROW_AUTOMATION_WORKFLOW_TIMEOUT_HOURS:
   BASEROW_AUTOMATION_WORKFLOW_HISTORY_MAX_DAYS:
@@ -338,6 +339,7 @@ services:
       BASEROW_AUTOMATION_WORKFLOW_RATE_LIMITS:
       BASEROW_AUTOMATION_WORKFLOW_RATE_LIMIT_CACHE_EXPIRY_SECONDS:
       BASEROW_AUTOMATION_WORKFLOW_HISTORY_RATE_LIMIT_CACHE_EXPIRY_SECONDS:
+      BASEROW_AUTOMATION_WORKFLOW_ERROR_LIMITS:
       BASEROW_AUTOMATION_WORKFLOW_MAX_CONSECUTIVE_ERRORS:
       BASEROW_AUTOMATION_WORKFLOW_TIMEOUT_HOURS:
       BASEROW_AUTOMATION_WORKFLOW_HISTORY_MAX_DAYS:
diff --git a/docs/installation/configuration.md b/docs/installation/configuration.md
index 8b7f7509e1..06bfdeb035 100644
--- a/docs/installation/configuration.md
+++ b/docs/installation/configuration.md
@@ -222,9 +222,10 @@ Baserow can throttle the number of concurrent requests a single user (or, option
 |---------------------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------|----------|
 | BASEROW\_AUTOMATION\_HISTORY\_PAGE\_SIZE\_LIMIT                     | The maximum number of automation history entries returned per page.                                                                          | 100      |
 | BASEROW\_AUTOMATION\_WORKFLOW\_RATE\_LIMIT\_MAX\_RUNS               | The maximum number of workflow runs that can be started within the rate limit window before new runs are blocked.                           | 10       |
-| BASEROW\_AUTOMATION\_WORKFLOW\_RATE\_LIMITS                         | A comma-separated list of integer pairs formatted as `max_runs,window_seconds`. For example, `10,5,20,60` means 10 executions per 5 seconds and 20 executions per 60 seconds. | empty    |
+| BASEROW\_AUTOMATION\_WORKFLOW\_RATE\_LIMITS                         | A comma-separated list of integer pairs formatted as `max_runs,window_seconds`. For example, `10,5,20,60` means 10 executions per 5 seconds and 20 executions per 60 seconds. | "10,5,30,300,100,3600"    |
 | BASEROW\_AUTOMATION\_WORKFLOW\_RATE\_LIMIT\_CACHE\_EXPIRY\_SECONDS  | The number of seconds the workflow rate limit counters are retained in cache.                                                               | 5        |
 | BASEROW\_AUTOMATION\_WORKFLOW\_HISTORY\_RATE\_LIMIT\_CACHE\_EXPIRY\_SECONDS | The number of seconds the workflow history rate limit counters are retained in cache. If unset, it uses the workflow rate limit cache expiry. | 5        |
+| BASEROW\_AUTOMATION\_WORKFLOW\_ERROR\_LIMITS                        | A comma-separated list of integer pairs formatted as `max_errors,window_seconds`. For example, `5,30,10,300` means more than 5 errors in 30 seconds or more than 10 errors in 300 seconds disables the workflow. | '20,300'    |
 | BASEROW\_AUTOMATION\_WORKFLOW\_MAX\_CONSECUTIVE\_ERRORS             | The maximum number of consecutive workflow errors allowed before the workflow is disabled.                                                  | 5        |
 | BASEROW\_AUTOMATION\_WORKFLOW\_TIMEOUT\_HOURS                       | The number of hours after which a running workflow is considered timed out.                                                                 | 24       |
 | BASEROW\_AUTOMATION\_WORKFLOW\_HISTORY\_MAX\_DAYS                   | The number of days automation workflow history entries are retained.                                                                         | 30       |
diff --git a/enterprise/web-frontend/modules/baserow_enterprise/components/teams/CreateTeamModal.vue b/enterprise/web-frontend/modules/baserow_enterprise/components/teams/CreateTeamModal.vue
index da22db3c45..635ab0b1eb 100644
--- a/enterprise/web-frontend/modules/baserow_enterprise/components/teams/CreateTeamModal.vue
+++ b/enterprise/web-frontend/modules/baserow_enterprise/components/teams/CreateTeamModal.vue
@@ -16,7 +16,7 @@
     <MemberAssignmentModal
       ref="memberAssignmentModal"
       :members="uninvitedUserSubjects"
-      @invite="storeSelectedUsers"
+      @select="storeSelectedUsers"
     />
   </Modal>
 </template>
diff --git a/enterprise/web-frontend/modules/baserow_enterprise/components/teams/UpdateTeamModal.vue b/enterprise/web-frontend/modules/baserow_enterprise/components/teams/UpdateTeamModal.vue
index ec7edd77a8..6ce94bdc84 100644
--- a/enterprise/web-frontend/modules/baserow_enterprise/components/teams/UpdateTeamModal.vue
+++ b/enterprise/web-frontend/modules/baserow_enterprise/components/teams/UpdateTeamModal.vue
@@ -20,7 +20,7 @@
     <MemberAssignmentModal
       ref="memberAssignmentModal"
       :members="uninvitedUserSubjects"
-      @invite="storeSelectedUsers"
+      @select="storeSelectedUsers"
     />
   </Modal>
 </template>
diff --git a/web-frontend/locales/en.json b/web-frontend/locales/en.json
index 5f2c6361e1..33c1eef2a6 100644
--- a/web-frontend/locales/en.json
+++ b/web-frontend/locales/en.json
@@ -49,6 +49,7 @@
     "cancel": "Cancel",
     "save": "Save",
     "set": "Set",
+    "select": "Select",
     "retry": "Retry",
     "search": "Search",
     "copy": "Copy",
diff --git a/web-frontend/modules/automation/components/AutomationHeader.vue b/web-frontend/modules/automation/components/AutomationHeader.vue
index ada8ca005a..6c8484977c 100644
--- a/web-frontend/modules/automation/components/AutomationHeader.vue
+++ b/web-frontend/modules/automation/components/AutomationHeader.vue
@@ -126,9 +126,10 @@
         </Button>
       </div>
     </div>
-    <AutomationSettingsModal
-      ref="automationSettingsModal"
+    <WorkflowSettingsModal
+      ref="workflowSettingsModal"
       :automation="automation"
+      :workflow="workflow"
     />
   </header>
 </template>
@@ -137,17 +138,17 @@
 import { useStore } from 'vuex'
 import moment from '@baserow/modules/core/moment'
 import { getUserTimeZone } from '@baserow/modules/core/utils/date'
-import { defineComponent, ref, computed } from 'vue'
+import { defineComponent, ref, computed, inject } from 'vue'
 import { HistoryEditorSidePanelType } from '@baserow/modules/automation/editorSidePanelTypes'
 import { notifyIf } from '@baserow/modules/core/utils/error'
 import { WORKFLOW_STATES } from '@baserow/modules/automation/components/enums'
 
 import NodeGraphHandler from '@baserow/modules/automation/utils/nodeGraphHandler'
-import AutomationSettingsModal from '@baserow/modules/automation/components/settings/AutomationSettingsModal'
+import WorkflowSettingsModal from '@baserow/modules/automation/components/settings/WorkflowSettingsModal'
 
 export default defineComponent({
   name: 'AutomationHeader',
-  components: { AutomationSettingsModal },
+  components: { WorkflowSettingsModal },
   props: {
     automation: {
       type: Object,
@@ -283,14 +284,15 @@ export default defineComponent({
       isPublishing.value = false
     }
 
-    const automationSettingsModal = ref(null)
+    const workflowSettingsModal = ref(null)
     const openSettingsModal = () => {
-      automationSettingsModal.value.show()
+      workflowSettingsModal.value.show()
     }
 
     return {
       isDev,
       debug,
+      workflow,
       statusSwitch,
       debugClick,
       historyClick,
@@ -303,11 +305,10 @@ export default defineComponent({
       isPublishing,
       isPaused,
       isDisabled,
-      workflow,
       activeSidePanel,
       testRunDisabled,
       openSettingsModal,
-      automationSettingsModal,
+      workflowSettingsModal,
     }
   },
 })
diff --git a/web-frontend/modules/automation/components/notifications/WorkflowDisabledNotification.vue b/web-frontend/modules/automation/components/notifications/WorkflowDisabledNotification.vue
new file mode 100644
index 0000000000..92781dc75d
--- /dev/null
+++ b/web-frontend/modules/automation/components/notifications/WorkflowDisabledNotification.vue
@@ -0,0 +1,30 @@
+<template>
+  <nuxt-link
+    class="notification-panel__notification-link"
+    :to="route"
+    @click="markAsReadAndHandleClick"
+  >
+    <div class="notification-panel__notification-content-title">
+      <i18n-t keypath="workflowDisabledNotification.body" tag="span">
+        <template #name>
+          <strong>{{ notification.data.workflow_name }}</strong>
+        </template>
+      </i18n-t>
+    </div>
+  </nuxt-link>
+</template>
+
+<script>
+import notificationContent from '@baserow/modules/core/mixins/notificationContent'
+
+export default {
+  name: 'WorkflowDisabledNotification',
+  mixins: [notificationContent],
+  emits: ['close-panel'],
+  methods: {
+    handleClick() {
+      this.$emit('close-panel')
+    },
+  },
+}
+</script>
diff --git a/web-frontend/modules/automation/components/settings/GeneralSettings.vue b/web-frontend/modules/automation/components/settings/GeneralSettings.vue
index 6ed92ca088..d9d7e37ffc 100644
--- a/web-frontend/modules/automation/components/settings/GeneralSettings.vue
+++ b/web-frontend/modules/automation/components/settings/GeneralSettings.vue
@@ -51,7 +51,9 @@ export default {
     },
   },
   setup() {
-    return { v$: useVuelidate({ $lazy: true }) }
+    return {
+      v$: useVuelidate({ $lazy: true }),
+    }
   },
   data() {
     return {
diff --git a/web-frontend/modules/automation/components/settings/WorkflowGeneralSettings.vue b/web-frontend/modules/automation/components/settings/WorkflowGeneralSettings.vue
new file mode 100644
index 0000000000..a2c34fa398
--- /dev/null
+++ b/web-frontend/modules/automation/components/settings/WorkflowGeneralSettings.vue
@@ -0,0 +1,213 @@
+<template>
+  <div ref="root">
+    <h2 class="box__title">
+      {{ $t('workflowGeneralSettings.titleOverview') }}
+    </h2>
+
+    <FormGroup
+      small-label
+      :label="$t('workflowGeneralSettings.nameLabel')"
+      :error-message="getFirstErrorMessage('name')"
+      required
+    >
+      <FormInput v-model="v$.values.name.$model" size="large"></FormInput>
+    </FormGroup>
+
+    <FormGroup
+      small-label
+      :label="$t('workflowGeneralSettings.workflowDisabledRecipientsLabel')"
+      class="padding-top-2"
+      :helper-text="
+        $t('workflowGeneralSettings.workflowDisabledRecipientsHelp')
+      "
+      required
+    >
+      <div v-if="selectedWorkflowRecipients.length > 0">
+        <Badge
+          v-for="recipient in selectedWorkflowRecipients"
+          :key="recipient.id"
+          class="margin-right-1 margin-top-1"
+          rounded
+          :title="recipient.email"
+        >
+          {{ recipient.name }}
+        </Badge>
+      </div>
+      <div v-else class="margin-top-2">
+        {{ $t('workflowGeneralSettings.noWorkflowDisabledRecipients') }}
+      </div>
+      <div class="margin-top-1">
+        <Button type="secondary" @click="openRecipientModal">
+          {{ $t('workflowGeneralSettings.selectWorkflowDisabledRecipients') }}
+        </Button>
+      </div>
+    </FormGroup>
+
+    <MemberAssignmentModal
+      ref="workflowDisabledRecipientsModal"
+      :members="workspaceMembers"
+      :selected-members="selectedWorkspaceMembers"
+      :allow-empty-selection="true"
+      :button-label="$t('action.select')"
+      @select="updateWorkflowDisabledRecipients"
+    />
+  </div>
+</template>
+
+<script setup>
+import { computed, onBeforeUnmount, onMounted, reactive, ref, toRef } from 'vue'
+import { useStore } from 'vuex'
+import { useNuxtApp } from '#imports'
+import { useI18n } from 'vue-i18n'
+import { useVuelidate } from '@vuelidate/core'
+import { maxLength, required, helpers } from '@vuelidate/validators'
+import debounce from 'lodash/debounce'
+import { useForm } from '@baserow/modules/core/composables/useForm'
+import WorkspaceService from '@baserow/modules/core/services/workspace'
+import { notifyIf } from '@baserow/modules/core/utils/error'
+import MemberAssignmentModal from '@baserow/modules/core/components/workspace/MemberAssignmentModal'
+import { clone } from '@baserow/modules/core/utils/object'
+
+defineOptions({
+  name: 'WorkflowGeneralSettings',
+})
+
+const props = defineProps({
+  automation: {
+    type: Object,
+    required: true,
+  },
+  workflow: {
+    type: Object,
+    required: true,
+  },
+  defaultValues: {
+    type: Object,
+    required: false,
+    default: () => {
+      return {}
+    },
+  },
+  disabled: {
+    type: Boolean,
+    required: false,
+    default: false,
+  },
+})
+
+const emit = defineEmits(['submitted', 'values-changed'])
+
+const { $client } = useNuxtApp()
+const { t } = useI18n()
+const store = useStore()
+const root = ref(null)
+const workflowDisabledRecipientsModal = ref(null)
+const workspaceMembers = ref([])
+const membersLoading = ref(false)
+const values = reactive({
+  name: '',
+  notification_recipient_ids: [],
+})
+const allowedValues = ['name', 'notification_recipient_ids']
+
+const rules = computed(() => ({
+  values: {
+    name: {
+      required: helpers.withMessage(t('error.requiredField'), required),
+      maxLength: helpers.withMessage(
+        t('error.maxLength', { max: 255 }),
+        maxLength(255)
+      ),
+    },
+  },
+}))
+const v$ = useVuelidate(rules, { values }, { $lazy: true })
+
+const loadWorkspaceMembers = async () => {
+  if (membersLoading.value || workspaceMembers.value.length) {
+    return
+  }
+
+  try {
+    membersLoading.value = true
+    const { data } = await WorkspaceService($client).fetchAllUsers(
+      props.automation.workspace.id
+    )
+    workspaceMembers.value = data
+  } catch (error) {
+    notifyIf(error, 'workspace')
+  } finally {
+    membersLoading.value = false
+  }
+}
+
+const openRecipientModal = async () => {
+  await loadWorkspaceMembers()
+  workflowDisabledRecipientsModal.value?.show()
+}
+
+onMounted(() => {
+  loadWorkspaceMembers()
+})
+
+const selectedWorkflowRecipientIds = computed(
+  () => values.notification_recipient_ids || []
+)
+const selectedWorkspaceMembers = computed(() =>
+  workspaceMembers.value.filter((member) =>
+    selectedWorkflowRecipientIds.value.includes(member.user_id)
+  )
+)
+const selectedWorkflowRecipients = computed(() =>
+  selectedWorkspaceMembers.value.map(({ user_id, name, email }) => ({
+    id: user_id,
+    name,
+    email,
+  }))
+)
+
+const debouncedUpdateWorkflow = debounce((updatedValues) => {
+  updateWorkflow(updatedValues)
+}, 300)
+
+const form = useForm({
+  defaultValues: toRef(props, 'defaultValues'),
+  values,
+  allowedValues,
+  emit,
+  v$,
+  root,
+  emitChange(newValues) {
+    debouncedUpdateWorkflow(newValues)
+  },
+})
+
+const updateWorkflow = async (updatedValues) => {
+  if (!form.isFormValid()) {
+    return
+  }
+
+  try {
+    await store.dispatch('automationWorkflow/update', {
+      automation: props.automation,
+      workflow: props.workflow,
+      values: clone(updatedValues),
+    })
+  } catch (error) {
+    notifyIf(error, 'automationWorkflow')
+    await form.reset()
+  }
+}
+
+const getFirstErrorMessage = form.getFirstErrorMessage
+
+const updateWorkflowDisabledRecipients = (membersSelected) => {
+  values.notification_recipient_ids = membersSelected.map(
+    ({ user_id: userId }) => userId
+  )
+}
+
+onBeforeUnmount(() => {
+  debouncedUpdateWorkflow.cancel()
+})
+</script>
diff --git a/web-frontend/modules/automation/components/settings/WorkflowSettingsModal.vue b/web-frontend/modules/automation/components/settings/WorkflowSettingsModal.vue
new file mode 100644
index 0000000000..9416b68cd7
--- /dev/null
+++ b/web-frontend/modules/automation/components/settings/WorkflowSettingsModal.vue
@@ -0,0 +1,30 @@
+<template>
+  <Modal ref="modal">
+    <WorkflowGeneralSettings
+      :automation="automation"
+      :workflow="workflow"
+      :default-values="workflow"
+    />
+  </Modal>
+</template>
+
+<script>
+import modal from '@baserow/modules/core/mixins/modal'
+import WorkflowGeneralSettings from '@baserow/modules/automation/components/settings/WorkflowGeneralSettings'
+
+export default {
+  name: 'WorkflowSettingsModal',
+  components: { WorkflowGeneralSettings },
+  mixins: [modal],
+  props: {
+    automation: {
+      type: Object,
+      required: true,
+    },
+    workflow: {
+      type: Object,
+      required: true,
+    },
+  },
+}
+</script>
diff --git a/web-frontend/modules/automation/locales/en.json b/web-frontend/modules/automation/locales/en.json
index 5b2cd3396d..3cbee576ba 100644
--- a/web-frontend/modules/automation/locales/en.json
+++ b/web-frontend/modules/automation/locales/en.json
@@ -60,6 +60,19 @@
     "cantUpdateAutomationTitle": "Couldn't Update Automation",
     "cantUpdateAutomationDescription": "Sorry, could not update the automation."
   },
+  "workflowGeneralSettings": {
+    "titleOverview": "Workflow settings",
+    "nameLabel": "Workflow name",
+    "workflowDisabledRecipientsLabel": "Notification recipients",
+    "workflowDisabledRecipientsHelp": "These workspace members will get an in-app notification if this workflow is automatically disabled.",
+    "selectWorkflowDisabledRecipients": "Select recipients",
+    "noWorkflowDisabledRecipients": "No recipients selected.",
+    "cantUpdateWorkflowTitle": "Couldn't Update Workflow",
+    "cantUpdateWorkflowDescription": "Sorry, could not update the workflow."
+  },
+  "workflowDisabledNotification": {
+    "body": "The {name} workflow was disabled."
+  },
   "integrationSettings": {
     "title": "Integrations",
     "noIntegrationMessage": "You have not yet created any integrations. They can be created by adding data source, action or user authentication.",
diff --git a/web-frontend/modules/automation/notificationTypes.jsx b/web-frontend/modules/automation/notificationTypes.jsx
new file mode 100644
index 0000000000..8e9c27b332
--- /dev/null
+++ b/web-frontend/modules/automation/notificationTypes.jsx
@@ -0,0 +1,34 @@
+import { NotificationType } from '@baserow/modules/core/notificationTypes'
+import Icon from '@baserow/modules/core/components/Icon'
+import WorkflowDisabledNotification from '@baserow/modules/automation/components/notifications/WorkflowDisabledNotification'
+
+export class WorkflowDisabledNotificationType extends NotificationType {
+  static getType() {
+    return 'automation_workflow_disabled'
+  }
+
+  getIconComponent() {
+    return () => (
+      <Icon
+        class="notification-panel__notification-automation-icon"
+        icon="baserow-icon-automation"
+        type="secondary"
+        size="large"
+      />
+    )
+  }
+
+  getContentComponent() {
+    return WorkflowDisabledNotification
+  }
+
+  getRoute(notificationData) {
+    return {
+      name: 'automation-workflow',
+      params: {
+        automationId: notificationData.automation_id,
+        workflowId: notificationData.workflow_id,
+      },
+    }
+  }
+}
diff --git a/web-frontend/modules/automation/plugin.js b/web-frontend/modules/automation/plugin.js
index 79f41f8a40..7bb7439bed 100644
--- a/web-frontend/modules/automation/plugin.js
+++ b/web-frontend/modules/automation/plugin.js
@@ -32,6 +32,7 @@ import {
   DuplicateAutomationWorkflowJobType,
   PublishAutomationWorkflowJobType,
 } from '@baserow/modules/automation/jobTypes'
+import { WorkflowDisabledNotificationType } from '@baserow/modules/automation/notificationTypes.jsx'
 import {
   HistoryEditorSidePanelType,
   NodeEditorSidePanelType,
@@ -117,6 +118,10 @@ export default defineNuxtPlugin({
     // Automation job types
     $registry.register('job', new DuplicateAutomationWorkflowJobType(context))
     $registry.register('job', new PublishAutomationWorkflowJobType(context))
+    $registry.register(
+      'notification',
+      new WorkflowDisabledNotificationType(context)
+    )
 
     // Automation settings
     $registry.registerNamespace('automationSettings')
diff --git a/web-frontend/modules/automation/store/automationWorkflow.js b/web-frontend/modules/automation/store/automationWorkflow.js
index 35682a09ee..58b7046509 100644
--- a/web-frontend/modules/automation/store/automationWorkflow.js
+++ b/web-frontend/modules/automation/store/automationWorkflow.js
@@ -125,7 +125,7 @@ const actions = {
   },
   async create({ commit, dispatch }, { automation, name }) {
     const { data: workflow } = await AutomationWorkflowService(
-      useNuxtApp().$client
+      this.$client
     ).create(automation.id, name)
 
     commit('ADD_ITEM', { automation, workflow })
@@ -135,7 +135,7 @@ const actions = {
     return workflow
   },
   async fetchById({ getters, commit, dispatch }, { automation, workflowId }) {
-    const { data } = await AutomationWorkflowService(useNuxtApp().$client).read(
+    const { data } = await AutomationWorkflowService(this.$client).read(
       workflowId
     )
     const workflow = getters.getById(automation, workflowId)
@@ -150,9 +150,10 @@ const actions = {
     return data
   },
   async update({ dispatch }, { automation, workflow, values }) {
-    const { data } = await AutomationWorkflowService(
-      useNuxtApp().$client
-    ).update(workflow.id, values)
+    const { data } = await AutomationWorkflowService(this.$client).update(
+      workflow.id,
+      values
+    )
 
     const update = Object.keys(values).reduce((result, key) => {
       result[key] = data[key]
@@ -160,9 +161,10 @@ const actions = {
     }, {})
 
     await dispatch('forceUpdate', { workflow, values: update })
+    return data
   },
   async delete({ dispatch }, { automation, workflow }) {
-    await AutomationWorkflowService(useNuxtApp().$client).delete(workflow.id)
+    await AutomationWorkflowService(this.$client).delete(workflow.id)
 
     await dispatch('forceDelete', { automation, workflow })
   },
@@ -173,10 +175,7 @@ const actions = {
     commit('ORDER_WORKFLOWS', { automation, order, isHashed })
 
     try {
-      await AutomationWorkflowService(useNuxtApp().$client).order(
-        automation.id,
-        order
-      )
+      await AutomationWorkflowService(this.$client).order(automation.id, order)
     } catch (error) {
       commit('ORDER_WORKFLOWS', { automation, order: oldOrder, isHashed })
       throw error
@@ -184,7 +183,7 @@ const actions = {
   },
   async duplicate({ commit, dispatch }, { workflow }) {
     const { data: job } = await AutomationWorkflowService(
-      useNuxtApp().$client
+      this.$client
     ).duplicate(workflow.id)
 
     await dispatch('job/create', job, { root: true })
@@ -195,10 +194,10 @@ const actions = {
     commit('SET_ACTIVE_SIDE_PANEL', sidePanelType)
   },
   async testRun({ dispatch }, { workflow }) {
-    await AutomationWorkflowService(useNuxtApp().$client).testRun(workflow.id)
+    await AutomationWorkflowService(this.$client).testRun(workflow.id)
   },
   async publishWorkflow({ dispatch }, { workflow }) {
-    await AutomationWorkflowService(useNuxtApp().$client).publish(workflow.id)
+    await AutomationWorkflowService(this.$client).publish(workflow.id)
   },
 }
 
diff --git a/web-frontend/modules/core/assets/scss/components/notification_panel.scss b/web-frontend/modules/core/assets/scss/components/notification_panel.scss
index a8abfe1b2f..cdf52a6a9b 100644
--- a/web-frontend/modules/core/assets/scss/components/notification_panel.scss
+++ b/web-frontend/modules/core/assets/scss/components/notification_panel.scss
@@ -144,6 +144,10 @@
   @include center-text(24px, 15px);
 }
 
+.notification-panel__notification-automation-icon {
+  margin-left: 3px;
+}
+
 .notification-panel__body {
   .infinite-scroll {
     top: 53px;
diff --git a/web-frontend/modules/core/components/workspace/MemberAssignmentModal.vue b/web-frontend/modules/core/components/workspace/MemberAssignmentModal.vue
index 33924c83bd..00a1366436 100644
--- a/web-frontend/modules/core/components/workspace/MemberAssignmentModal.vue
+++ b/web-frontend/modules/core/components/workspace/MemberAssignmentModal.vue
@@ -4,7 +4,10 @@
       ref="memberSelectionList"
       class="padding-top-2"
       :members="members"
-      @invite="storeSelectedMembers"
+      :selected-members="selectedMembers"
+      :allow-empty-selection="allowEmptySelection"
+      :button-label="buttonLabel"
+      @select="storeSelectedMembers"
     />
   </Modal>
 </template>
@@ -22,11 +25,26 @@ export default {
       type: Array,
       required: true,
     },
+    selectedMembers: {
+      type: Array,
+      required: false,
+      default: () => [],
+    },
+    allowEmptySelection: {
+      type: Boolean,
+      required: false,
+      default: false,
+    },
+    buttonLabel: {
+      type: String,
+      required: false,
+      default: null,
+    },
   },
-  emits: ['invite'],
+  emits: ['select'],
   methods: {
     storeSelectedMembers(membersSelected) {
-      this.$emit('invite', membersSelected)
+      this.$emit('select', membersSelected)
       this.hide()
     },
   },
diff --git a/web-frontend/modules/core/components/workspace/MemberAssignmentModalFooter.vue b/web-frontend/modules/core/components/workspace/MemberAssignmentModalFooter.vue
index e6f3635ea0..2d785859ef 100644
--- a/web-frontend/modules/core/components/workspace/MemberAssignmentModalFooter.vue
+++ b/web-frontend/modules/core/components/workspace/MemberAssignmentModalFooter.vue
@@ -11,11 +11,9 @@
     <div class="col col-6 align-right">
       <Button
         type="primary"
-        :disabled="!inviteEnabled"
-        @click="inviteEnabled ? $emit('invite') : null"
-        >{{
-          $t('memberAssignmentModalFooter.invite', { selectedMembersCount })
-        }}
+        :disabled="!selectEnabled"
+        @click="selectEnabled ? $emit('select') : null"
+        >{{ actionLabel }}
       </Button>
     </div>
   </div>
@@ -38,14 +36,32 @@ export default {
       required: false,
       default: false,
     },
+    buttonLabel: {
+      type: String,
+      required: false,
+      default: null,
+    },
+    allowEmptySelection: {
+      type: Boolean,
+      required: false,
+      default: false,
+    },
   },
-  emits: ['invite', 'toggle-select-all'],
+  emits: ['select', 'toggle-select-all'],
   computed: {
     toggleEnabled() {
       return this.filteredMembersCount !== 0
     },
-    inviteEnabled() {
-      return this.selectedMembersCount !== 0
+    selectEnabled() {
+      return this.allowEmptySelection || this.selectedMembersCount !== 0
+    },
+    actionLabel() {
+      return (
+        this.buttonLabel ??
+        this.$t('memberAssignmentModalFooter.invite', {
+          selectedMembersCount: this.selectedMembersCount,
+        })
+      )
     },
     getToggleLabel() {
       return this.allFilteredMembersSelected
diff --git a/web-frontend/modules/core/components/workspace/MemberSelectionList.vue b/web-frontend/modules/core/components/workspace/MemberSelectionList.vue
index fdc0f5474a..ffe28fe371 100644
--- a/web-frontend/modules/core/components/workspace/MemberSelectionList.vue
+++ b/web-frontend/modules/core/components/workspace/MemberSelectionList.vue
@@ -39,8 +39,10 @@
       :all-filtered-members-selected="allFilteredMembersSelected"
       :selected-members-count="membersSelected.length"
       :filtered-members-count="membersFiltered.length"
+      :button-label="buttonLabel"
+      :allow-empty-selection="allowEmptySelection"
       @toggle-select-all="toggleSelectAll"
-      @invite="$emit('invite', membersSelected)"
+      @select="$emit('select', membersSelected)"
     />
   </div>
 </template>
@@ -55,12 +57,27 @@ export default {
       type: Array,
       required: true,
     },
+    selectedMembers: {
+      type: Array,
+      required: false,
+      default: () => [],
+    },
+    allowEmptySelection: {
+      type: Boolean,
+      required: false,
+      default: false,
+    },
+    buttonLabel: {
+      type: String,
+      required: false,
+      default: null,
+    },
   },
-  emits: ['invite'],
+  emits: ['select'],
   data() {
     return {
       membersFiltered: this.members,
-      membersSelected: [],
+      membersSelected: [...this.selectedMembers],
       activeSearchTerm: null,
     }
   },
@@ -79,6 +96,12 @@ export default {
     activeSearchTerm(newValue) {
       this.search(newValue)
     },
+    members(newValue) {
+      this.membersFiltered = newValue
+    },
+    selectedMembers(newValue) {
+      this.membersSelected = [...newValue]
+    },
   },
   mounted() {
     this.$priorityBus.$on(
diff --git a/web-frontend/modules/core/composables/useForm.js b/web-frontend/modules/core/composables/useForm.js
new file mode 100644
index 0000000000..80698d625d
--- /dev/null
+++ b/web-frontend/modules/core/composables/useForm.js
@@ -0,0 +1,241 @@
+import {
+  inject,
+  nextTick,
+  onBeforeUnmount,
+  provide,
+  ref,
+  unref,
+  watch,
+} from 'vue'
+import get from 'lodash/get'
+
+import { clone } from '@baserow/modules/core/utils/object'
+
+const formParentKey = Symbol('formParentKey')
+
+export function useForm({
+  defaultValues,
+  values,
+  allowedValues = null,
+  emit,
+  v$ = null,
+  emitChange = null,
+  root = null,
+  selectedFieldIsDeactivated = false,
+}) {
+  const parentForm = inject(formParentKey, null)
+  const emitValues = ref(true)
+  const skipFirstValuesEmit = ref(false)
+  const registeredChildForms = ref([])
+
+  const getVuelidateField = (fieldName) => {
+    const vuelidate = unref(v$)
+    if (!vuelidate) {
+      return null
+    }
+
+    return (
+      get(vuelidate, fieldName) ?? get(vuelidate, `values.${fieldName}`) ?? null
+    )
+  }
+
+  const isAllowedKey = (key) => {
+    const currentAllowedValues = unref(allowedValues)
+    if (currentAllowedValues !== null) {
+      return currentAllowedValues.includes(key)
+    }
+    return true
+  }
+
+  const getCurrentDefaultValues = () => {
+    const currentDefaultValues = unref(defaultValues) ?? {}
+
+    return Object.keys(currentDefaultValues).reduce((result, key) => {
+      if (isAllowedKey(key)) {
+        let value = currentDefaultValues[key]
+
+        if (
+          Array.isArray(value) ||
+          (typeof value === 'object' && value !== null)
+        ) {
+          value = clone(value)
+        }
+
+        result[key] = value
+      }
+      return result
+    }, {})
+  }
+
+  Object.assign(values, getCurrentDefaultValues())
+
+  const formApi = {
+    emitValues,
+    skipFirstValuesEmit,
+    registeredChildForms,
+    registerChildForm(childForm) {
+      if (!registeredChildForms.value.includes(childForm)) {
+        registeredChildForms.value.push(childForm)
+      }
+    },
+    unregisterChildForm(childForm) {
+      const index = registeredChildForms.value.indexOf(childForm)
+      if (index !== -1) {
+        registeredChildForms.value.splice(index, 1)
+      }
+    },
+    isAllowedKey,
+    getDefaultValues: getCurrentDefaultValues,
+    focusOnFirstError() {
+      const element = unref(root)?.$el ?? unref(root)
+      const firstError = element?.querySelector?.('[data-form-error]')
+      if (firstError) {
+        firstError.scrollIntoView({ behavior: 'smooth' })
+      }
+    },
+    getChildForms(predicate = (child) => 'isFormValid' in child, deep = false) {
+      const children = []
+
+      const processChildren = (forms, depth = 0) => {
+        for (const form of unref(forms) ?? []) {
+          if (predicate(form)) {
+            children.push(form)
+          }
+
+          if (deep && depth < 10 && form.registeredChildForms) {
+            processChildren(form.registeredChildForms, depth + 1)
+          }
+        }
+      }
+
+      processChildren(registeredChildForms.value)
+      return children
+    },
+    touch(deep = false) {
+      unref(v$)?.$touch()
+
+      for (const child of formApi.getChildForms(
+        (child) => 'touch' in child,
+        deep
+      )) {
+        child.touch(deep)
+      }
+    },
+    submit(deep = false) {
+      if (unref(selectedFieldIsDeactivated)) {
+        return
+      }
+
+      formApi.touch(deep)
+
+      if (formApi.isFormValid(deep)) {
+        emit?.('submitted', formApi.getFormValues(deep))
+      } else {
+        nextTick(() => formApi.focusOnFirstError())
+      }
+    },
+    fieldHasErrors(fieldName) {
+      return getVuelidateField(fieldName)?.$error || false
+    },
+    getFirstErrorMessage(fieldName) {
+      return getVuelidateField(fieldName)?.$errors?.[0]?.$message
+    },
+    isFormValid(deep = false) {
+      const thisFormInvalid = Boolean(unref(v$)?.$invalid)
+      return !thisFormInvalid && formApi.areChildFormsValid(deep)
+    },
+    areChildFormsValid(deep = false) {
+      return formApi
+        .getChildForms((child) => 'isFormValid' in child, deep)
+        .every((child) => child.isFormValid())
+    },
+    getFormValues(deep = false) {
+      return Object.assign({}, values, formApi.getChildFormsValues(deep))
+    },
+    getChildFormsValues(deep = false) {
+      const children = formApi.getChildForms(
+        (child) => 'getChildFormsValues' in child,
+        deep
+      )
+      return Object.assign(
+        {},
+        ...children.map((child) => child.getFormValues(deep))
+      )
+    },
+    isDirty() {
+      for (const [key, value] of Object.entries(getCurrentDefaultValues())) {
+        if (values[key] !== value) {
+          return true
+        }
+      }
+      return false
+    },
+    async reset(deep = false) {
+      Object.assign(values, getCurrentDefaultValues())
+
+      unref(v$)?.$reset()
+
+      await nextTick()
+
+      formApi
+        .getChildForms((child) => 'reset' in child, deep)
+        .forEach((child) => child.reset())
+    },
+    setEmitValues(value) {
+      emitValues.value = value
+      formApi
+        .getChildForms((child) => 'setEmitValues' in child, true)
+        .forEach((child) => child.setEmitValues(value))
+    },
+    handleErrorByForm(error, deep = false) {
+      let childHandledIt = false
+      const children = formApi.getChildForms(
+        (child) => 'handleErrorByForm' in child,
+        deep
+      )
+      for (const child of children) {
+        if (child.handleErrorByForm(error)) {
+          childHandledIt = true
+        }
+      }
+      return childHandledIt
+    },
+    emitChange(newValues) {
+      if (emitChange) {
+        emitChange(newValues)
+        return
+      }
+
+      emit?.('values-changed', newValues)
+    },
+  }
+
+  provide(formParentKey, formApi)
+
+  if (typeof parentForm?.registerChildForm === 'function') {
+    parentForm.registerChildForm(formApi)
+  }
+
+  onBeforeUnmount(() => {
+    if (typeof parentForm?.unregisterChildForm === 'function') {
+      parentForm.unregisterChildForm(formApi)
+    }
+  })
+
+  watch(
+    values,
+    (newValues) => {
+      if (skipFirstValuesEmit.value) {
+        skipFirstValuesEmit.value = false
+        return
+      }
+
+      if (emitValues.value) {
+        formApi.emitChange(newValues)
+      }
+    },
+    { deep: true }
+  )
+
+  return formApi
+}
diff --git a/web-frontend/test/unit/automation/notificationTypes.spec.js b/web-frontend/test/unit/automation/notificationTypes.spec.js
new file mode 100644
index 0000000000..7cd8f8ea8e
--- /dev/null
+++ b/web-frontend/test/unit/automation/notificationTypes.spec.js
@@ -0,0 +1,20 @@
+import { WorkflowDisabledNotificationType } from '@baserow/modules/automation/notificationTypes'
+
+describe('WorkflowDisabledNotificationType', () => {
+  test('resolves the workflow route from notification data', () => {
+    const type = new WorkflowDisabledNotificationType({ app: {} })
+
+    expect(
+      type.getRoute({
+        automation_id: 12,
+        workflow_id: 34,
+      })
+    ).toStrictEqual({
+      name: 'automation-workflow',
+      params: {
+        automationId: 12,
+        workflowId: 34,
+      },
+    })
+  })
+})
diff --git a/web-frontend/test/unit/core/composables/useForm.spec.js b/web-frontend/test/unit/core/composables/useForm.spec.js
new file mode 100644
index 0000000000..dfdceea99e
--- /dev/null
+++ b/web-frontend/test/unit/core/composables/useForm.spec.js
@@ -0,0 +1,85 @@
+import { nextTick, reactive, ref, toRef, defineComponent } from 'vue'
+import { mountSuspended } from '@nuxt/test-utils/runtime'
+
+import { useForm } from '@baserow/modules/core/composables/useForm'
+
+const TestForm = defineComponent({
+  props: {
+    defaultValues: {
+      type: Object,
+      default: () => ({}),
+    },
+  },
+  emits: ['submitted', 'values-changed'],
+  setup(props, { emit }) {
+    const values = reactive({
+      name: '',
+      notification_recipient_ids: [],
+      ignored: 'original',
+    })
+    const root = ref(null)
+
+    return {
+      root,
+      values,
+      ...useForm({
+        defaultValues: toRef(props, 'defaultValues'),
+        values,
+        allowedValues: ['name', 'notification_recipient_ids'],
+        emit,
+        root,
+      }),
+    }
+  },
+  template: '<div ref="root"></div>',
+})
+
+describe('useForm', () => {
+  it('hydrates allowed default values with cloned objects and emits changes', async () => {
+    const defaultValues = {
+      name: 'Workflow name',
+      notification_recipient_ids: [1, 2],
+      ignored: 'should not be applied',
+    }
+
+    const wrapper = await mountSuspended(TestForm, {
+      props: { defaultValues },
+    })
+
+    expect(wrapper.vm.values.name).toBe('Workflow name')
+    expect(wrapper.vm.values.notification_recipient_ids).toEqual([1, 2])
+    expect(wrapper.vm.values.notification_recipient_ids).not.toBe(
+      defaultValues.notification_recipient_ids
+    )
+    expect(wrapper.vm.values.ignored).toBe('original')
+
+    defaultValues.notification_recipient_ids.push(3)
+    expect(wrapper.vm.values.notification_recipient_ids).toEqual([1, 2])
+
+    wrapper.vm.values.name = 'Updated workflow'
+    await nextTick()
+
+    expect(wrapper.emitted('values-changed')[0][0].name).toBe(
+      'Updated workflow'
+    )
+  })
+
+  it('resets to the filtered default values', async () => {
+    const wrapper = await mountSuspended(TestForm, {
+      props: {
+        defaultValues: {
+          name: 'Initial workflow',
+          notification_recipient_ids: [5],
+        },
+      },
+    })
+
+    wrapper.vm.values.name = 'Changed workflow'
+    wrapper.vm.values.notification_recipient_ids = [9]
+
+    await wrapper.vm.reset()
+
+    expect(wrapper.vm.values.name).toBe('Initial workflow')
+    expect(wrapper.vm.values.notification_recipient_ids).toEqual([5])
+  })
+})