diff --git a/backend/src/baserow/contrib/automation/api/workflows/serializers.py b/backend/src/baserow/contrib/automation/api/workflows/serializers.py
index 21e1fc2928..b3ace055fc 100644
--- a/backend/src/baserow/contrib/automation/api/workflows/serializers.py
+++ b/backend/src/baserow/contrib/automation/api/workflows/serializers.py
@@ -124,8 +124,17 @@ class Meta:
         )
 
 
+class AutomationWorkflowHistoryTriggeredBySerializer(serializers.Serializer):
+    """Who started the run, as `{id, type, name}` for the collaborator UI."""
+
+    id = serializers.IntegerField(source="triggered_by_id", read_only=True)
+    type = serializers.CharField(source="triggered_by_type", read_only=True)
+    name = serializers.CharField(source="triggered_by_name", read_only=True)
+
+
 class AutomationWorkflowHistorySerializer(AutomationHistorySerializer):
     plugin_data = serializers.SerializerMethodField()
+    triggered_by = serializers.SerializerMethodField()
 
     class Meta:
         model = AutomationWorkflowHistory
@@ -133,12 +142,21 @@ class Meta:
             "is_test_run",
             "simulate_until_node",
             "plugin_data",
+            "triggered_by",
         )
 
     @extend_schema_field(serializers.DictField())
     def get_plugin_data(self, obj):
         return self.context.get("workflow_history_plugin_data", {}).get(obj.id, {})
 
+    @extend_schema_field(
+        AutomationWorkflowHistoryTriggeredBySerializer(allow_null=True)
+    )
+    def get_triggered_by(self, obj):
+        if obj.triggered_by_id is None:
+            return None
+        return AutomationWorkflowHistoryTriggeredBySerializer(obj).data
+
 
 class AutomationWorkflowHistoryPagination(PageNumberPagination):
     def get_paginated_response(self, data, *, success_count: int, fail_count: int):
diff --git a/backend/src/baserow/contrib/automation/automation_dispatch_context.py b/backend/src/baserow/contrib/automation/automation_dispatch_context.py
index c282f73a7f..8dc8a12902 100644
--- a/backend/src/baserow/contrib/automation/automation_dispatch_context.py
+++ b/backend/src/baserow/contrib/automation/automation_dispatch_context.py
@@ -4,9 +4,7 @@
     automation_data_provider_type_registry,
 )
 from baserow.contrib.automation.history.handler import AutomationHistoryHandler
-from baserow.contrib.automation.history.models import (
-    AutomationNodeHistory,
-)
+from baserow.contrib.automation.history.models import AutomationWorkflowHistory
 from baserow.contrib.automation.nodes.models import AutomationActionNode
 from baserow.contrib.automation.workflows.models import AutomationWorkflow
 from baserow.core.cache import local_cache
@@ -19,7 +17,7 @@ class AutomationDispatchContext(DispatchContext):
     def __init__(
         self,
         workflow: AutomationWorkflow,
-        history: AutomationNodeHistory,
+        history: AutomationWorkflowHistory,
         event_payload: Optional[Union[Dict, List[Dict]]] = None,
         simulate_until_node: Optional[AutomationActionNode] = None,
         current_iterations: Optional[Dict[int, int]] = None,
diff --git a/backend/src/baserow/contrib/automation/history/handler.py b/backend/src/baserow/contrib/automation/history/handler.py
index b38d6a140b..5704650f90 100644
--- a/backend/src/baserow/contrib/automation/history/handler.py
+++ b/backend/src/baserow/contrib/automation/history/handler.py
@@ -17,8 +17,10 @@
 from baserow.contrib.automation.nodes.models import AutomationNode
 from baserow.contrib.automation.workflows.models import AutomationWorkflow
 from baserow.core.db import specific_iterator
+from baserow.core.registries import subject_type_registry
 from baserow.core.services.handler import ServiceHandler
 from baserow.core.services.models import Service
+from baserow.core.types import Subject
 
 
 class AutomationHistoryHandler:
@@ -73,8 +75,25 @@ def create_workflow_history(
         status: HistoryStatusChoices = HistoryStatusChoices.STARTED,
         completed_on: Optional[datetime] = None,
         message: str = "",
+        triggered_by: Optional[Subject] = None,
     ) -> AutomationWorkflowHistory:
-        """Creates a history entry for a Workflow run."""
+        """
+        Creates a history entry for a Workflow run.
+
+        :param triggered_by: The subject who deliberately started the run, when
+            one did. An event-started run has none.
+        """
+
+        triggered_by_values = {}
+        if triggered_by is not None:
+            subject_type = subject_type_registry.get_by_model(triggered_by)
+            triggered_by_values = {
+                "triggered_by_id": triggered_by.id,
+                "triggered_by_type": subject_type.type,
+                # TODO: use `subject_type.get_display_name` once agents are
+                #  subjects (#6064). Users are the only subject starting a run.
+                "triggered_by_name": triggered_by.first_name,
+            }
 
         return AutomationWorkflowHistory.objects.create(
             workflow=workflow,
@@ -86,6 +105,7 @@ def create_workflow_history(
             status=status,
             completed_on=completed_on,
             message=message,
+            **triggered_by_values,
         )
 
     def create_node_history(
diff --git a/backend/src/baserow/contrib/automation/history/models.py b/backend/src/baserow/contrib/automation/history/models.py
index fca3cd715b..e0c4ee9a4e 100644
--- a/backend/src/baserow/contrib/automation/history/models.py
+++ b/backend/src/baserow/contrib/automation/history/models.py
@@ -1,6 +1,7 @@
 from django.db import models
 
 from baserow.contrib.automation.history.constants import HistoryStatusChoices
+from baserow.core.subjects import UserSubjectType
 
 
 class AutomationHistory(models.Model):
@@ -50,6 +51,27 @@ class AutomationWorkflowHistory(AutomationHistory):
         help_text="Event payload received by the workflow.",
     )
 
+    # Who started the run, as a subject rather than a user foreign key so an
+    # agent can be recorded too. The id is only meaningful together with the
+    # type, since user and agent ids can overlap. The name is kept so the entry
+    # still reads after the subject is deleted.
+    triggered_by_id = models.PositiveIntegerField(
+        null=True,
+        help_text="The id of the subject who started this run. Null when an "
+        "event started it.",
+    )
+    triggered_by_type = models.CharField(
+        max_length=255,
+        db_default=UserSubjectType.type,
+        help_text="The subject type of who started this run.",
+    )
+    triggered_by_name = models.CharField(
+        max_length=160,
+        blank=True,
+        db_default="",
+        help_text="The name of who started this run, when it started.",
+    )
+
     class Meta(AutomationHistory.Meta):
         indexes = [
             models.Index(
diff --git a/backend/src/baserow/contrib/automation/migrations/0036_triggered_by.py b/backend/src/baserow/contrib/automation/migrations/0036_triggered_by.py
new file mode 100644
index 0000000000..d37a6078a7
--- /dev/null
+++ b/backend/src/baserow/contrib/automation/migrations/0036_triggered_by.py
@@ -0,0 +1,38 @@
+# Generated by Django 5.2.16 on 2026-09-14 11:44
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('automation', '0035_coregotoactionnode'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='automationworkflow',
+            name='test_run_triggered_by_id',
+            field=models.PositiveIntegerField(blank=True, help_text='The id of the subject who asked for the pending test run.', null=True),
+        ),
+        migrations.AddField(
+            model_name='automationworkflow',
+            name='test_run_triggered_by_type',
+            field=models.CharField(db_default='auth.User', help_text='The subject type of who asked for the pending test run.', max_length=255),
+        ),
+        migrations.AddField(
+            model_name='automationworkflowhistory',
+            name='triggered_by_id',
+            field=models.PositiveIntegerField(help_text='The id of the subject who started this run. Null when an event started it.', null=True),
+        ),
+        migrations.AddField(
+            model_name='automationworkflowhistory',
+            name='triggered_by_name',
+            field=models.CharField(blank=True, db_default='', help_text='The name of who started this run, when it started.', max_length=160),
+        ),
+        migrations.AddField(
+            model_name='automationworkflowhistory',
+            name='triggered_by_type',
+            field=models.CharField(db_default='auth.User', help_text='The subject type of who started this run.', max_length=255),
+        ),
+    ]
diff --git a/backend/src/baserow/contrib/automation/nodes/node_types.py b/backend/src/baserow/contrib/automation/nodes/node_types.py
index cb3521346d..890b8d5243 100644
--- a/backend/src/baserow/contrib/automation/nodes/node_types.py
+++ b/backend/src/baserow/contrib/automation/nodes/node_types.py
@@ -702,6 +702,10 @@ def on_event(
             AutomationWorkflowHandler().async_start_workflow(
                 workflow,
                 service_payload,
+                # Read before the reset below clears it.
+                triggered_by=AutomationWorkflowHandler().get_test_run_triggered_by(
+                    workflow
+                ),
             )
 
             # We don't want subsequent events to trigger a new test run
diff --git a/backend/src/baserow/contrib/automation/workflows/handler.py b/backend/src/baserow/contrib/automation/workflows/handler.py
index 587e73d37d..f7d58bcba7 100644
--- a/backend/src/baserow/contrib/automation/workflows/handler.py
+++ b/backend/src/baserow/contrib/automation/workflows/handler.py
@@ -58,10 +58,12 @@
 from baserow.contrib.automation.workflows.types import UpdatedAutomationWorkflow
 from baserow.core.cache import global_cache, local_cache
 from baserow.core.exceptions import IdDoesNotExist
-from baserow.core.registries import ImportExportConfig
+from baserow.core.registries import ImportExportConfig, subject_type_registry
 from baserow.core.storage import ExportZipFile, get_default_storage
+from baserow.core.subjects import UserSubjectType
 from baserow.core.telemetry.utils import baserow_trace, baserow_trace_handler
 from baserow.core.trash.handler import TrashHandler
+from baserow.core.types import Subject
 from baserow.core.utils import (
     ChildProgressBuilder,
     MirrorDict,
@@ -228,12 +230,17 @@ def export_prepared_values(self, workflow: AutomationWorkflow) -> Dict[Any, Any]
         return prepared_values
 
     def update_workflow(
-        self, workflow: AutomationWorkflow, **kwargs
+        self,
+        workflow: AutomationWorkflow,
+        triggered_by: Optional[Subject] = None,
+        **kwargs,
     ) -> UpdatedAutomationWorkflow:
         """
         Updates fields of the provided AutomationWorkflow.
 
         :param workflow: The AutomationWorkflow that should be updated.
+        :param triggered_by: Who makes the update, recorded as the starter of
+            the test run when the update opens the test run window.
         :param kwargs: The fields that should be updated with their
             corresponding values.
         :return: The updated AutomationWorkflow.
@@ -254,9 +261,21 @@ def update_workflow(
                 published_workflow.state = WorkflowState(state)
                 published_workflow.save(update_fields=["state"])
 
+        previous_allow_test_run_until = workflow.allow_test_run_until
         for key, value in extract_allowed(allowed_values, attr_fields).items():
             setattr(workflow, key, value)
 
+        # Opening the window records who opened it; closing it forgets them, so a
+        # later run never names someone from an earlier window. Undo and redo
+        # send every field back, as JSON, so an unchanged window keeps its starter.
+        allow_test_run_until = AutomationWorkflow._meta.get_field(
+            "allow_test_run_until"
+        ).to_python(workflow.allow_test_run_until)
+        if allow_test_run_until != previous_allow_test_run_until:
+            self._set_test_run_triggered_by(
+                workflow, triggered_by if workflow.allow_test_run_until else None
+            )
+
         workflow.save()
         set_allowed_m2m_fields(allowed_values, m2m_fields, workflow)
 
@@ -756,7 +775,12 @@ def disable_workflow(self, workflow: AutomationWorkflow) -> None:
 
         automation_workflow_updated.send(self, user=None, workflow=original_workflow)
 
-    def set_workflow_temporary_states(self, workflow, simulate_until_node=None):
+    def set_workflow_temporary_states(
+        self,
+        workflow,
+        simulate_until_node=None,
+        triggered_by: Optional[Subject] = None,
+    ):
         """
         Sets the temporary states necessary to allow an unpublished workflow to be
         ran by the next event. By default a full test run is scheduled unless the
@@ -764,9 +788,13 @@ def set_workflow_temporary_states(self, workflow, simulate_until_node=None):
 
         :param workflow: The workflow to consider.
         :param simulate_until_node: If set, schedules a simulation run instead.
+        :param triggered_by: Who asked for the run, recorded on its history
+            even when the run waits for an event.
         """
 
-        fields_to_save = []
+        # Always written, so a run never names the starter of an earlier one.
+        fields_to_save = self._set_test_run_triggered_by(workflow, triggered_by)
+
         if simulate_until_node is not None:
             # Switch to simulate until the given node
             workflow.simulate_until_node = simulate_until_node
@@ -786,6 +814,44 @@ def set_workflow_temporary_states(self, workflow, simulate_until_node=None):
             workflow.save(update_fields=fields_to_save)
             automation_workflow_updated.send(self, user=None, workflow=workflow)
 
+    def _set_test_run_triggered_by(
+        self, workflow: AutomationWorkflow, triggered_by: Optional[Subject]
+    ) -> List[str]:
+        """
+        Sets who asked for the workflow's pending test run, without saving.
+
+        :param workflow: The workflow waiting for its test run.
+        :param triggered_by: The subject, or None to record nobody.
+        :return: The fields to save.
+        """
+
+        if triggered_by is None:
+            workflow.test_run_triggered_by_id = None
+            workflow.test_run_triggered_by_type = UserSubjectType.type
+            return ["test_run_triggered_by_id", "test_run_triggered_by_type"]
+
+        workflow.test_run_triggered_by_id = triggered_by.id
+        workflow.test_run_triggered_by_type = subject_type_registry.get_by_model(
+            triggered_by
+        ).type
+        return ["test_run_triggered_by_id", "test_run_triggered_by_type"]
+
+    def get_test_run_triggered_by(self, workflow) -> Optional[Subject]:
+        """
+        Returns who asked for the workflow's pending test run or simulation.
+
+        :param workflow: The workflow waiting for its test run.
+        :return: The subject, or None when nobody is recorded or it no longer
+            exists.
+        """
+
+        if workflow.test_run_triggered_by_id is None:
+            return None
+
+        return subject_type_registry.get_subject(
+            workflow.test_run_triggered_by_type, workflow.test_run_triggered_by_id
+        )
+
     def reset_workflow_temporary_states(self, workflow):
         """
         Reset the temporary states set when we want to test or simulate a workflow.
@@ -801,6 +867,9 @@ def reset_workflow_temporary_states(self, workflow):
             workflow.simulate_until_node = None
             fields_to_save.append("simulate_until_node")
 
+        if workflow.test_run_triggered_by_id is not None:
+            fields_to_save += self._set_test_run_triggered_by(workflow, None)
+
         if fields_to_save:
             workflow.save(update_fields=fields_to_save)
             automation_workflow_updated.send(self, user=None, workflow=workflow)
@@ -810,6 +879,7 @@ def toggle_test_run(
         self,
         workflow: AutomationWorkflow,
         simulate_until_node: AutomationNode | None,
+        triggered_by: Optional[AbstractUser] = None,
     ):
         """
         Trigger a test run if none is in progress or cancel the planned run. If the
@@ -820,6 +890,8 @@ def toggle_test_run(
 
         :param workflow: The workflow we want to trigger the test run for.
         :param simulate_until_node: If we want to simulate until a particular node.
+        :param triggered_by: The person starting the test run, recorded on its
+            history entry.
         """
 
         if workflow.simulate_until_node is not None or workflow.allow_test_run_until:
@@ -828,22 +900,27 @@ def toggle_test_run(
             return
 
         if simulate_until_node is None:  # Full test
-            AutomationWorkflowHandler().set_workflow_temporary_states(workflow)
+            AutomationWorkflowHandler().set_workflow_temporary_states(
+                workflow, triggered_by=triggered_by
+            )
             if workflow.can_be_immediately_dispatched():
                 # If the service related to the trigger can immediately dispatch,
                 # we immediately trigger the workflow run.
-                self.async_start_workflow(workflow)
+                self.async_start_workflow(workflow, triggered_by=triggered_by)
         else:
             AutomationWorkflowHandler().set_workflow_temporary_states(
-                workflow, simulate_until_node=simulate_until_node
+                workflow,
+                simulate_until_node=simulate_until_node,
+                triggered_by=triggered_by,
             )
             trigger = workflow.get_trigger()
 
             dispatch_context = AutomationDispatchContext(
                 workflow,
-                # This is a placeholder value, no actual history exists yet
-                # (it's created later in start_workflow). This is fine
-                # for now, because get_sample_data() doesn't use history.
+                # No history exists yet, start_workflow creates it. This context
+                # only serves the get_sample_data() call below, which never reads
+                # history, and is never used for a run. Every other context
+                # needs a real history.
                 history=None,
                 simulate_until_node=simulate_until_node,
             )
@@ -857,7 +934,7 @@ def toggle_test_run(
                 # If the trigger is immediately dispatchable or if we already have
                 # the sample data for it we can immediately dispatch the workflow
                 # except if we are updating the trigger sample data by itself
-                self.async_start_workflow(workflow)
+                self.async_start_workflow(workflow, triggered_by=triggered_by)
 
     @baserow_trace(tracer)
     def clear_old_history(self) -> None:
@@ -1143,12 +1220,15 @@ def async_start_workflow(
         self,
         workflow: AutomationWorkflow,
         event_payload: Optional[List[Dict]] = None,
+        triggered_by: Optional[AbstractUser] = None,
     ) -> None:
         """
         Runs the provided workflow in a celery task.
 
         :param workflow: The AutomationWorkflow ID that should be executed.
         :param event_payload: The payload from the action.
+        :param triggered_by: The person who started the run, recorded on the
+            history entry.
         """
 
         error = None
@@ -1216,6 +1296,7 @@ def async_start_workflow(
                     completed_on=now,
                     message=error,
                     status=history_status,
+                    triggered_by=triggered_by,
                 )
             return
 
@@ -1226,6 +1307,7 @@ def async_start_workflow(
             is_test_run=is_test_run,
             event_payload=event_payload,
             simulate_until_node=simulate_until_node,
+            triggered_by=triggered_by,
         )
 
         automation_workflow_dispatch_started.send(
diff --git a/backend/src/baserow/contrib/automation/workflows/models.py b/backend/src/baserow/contrib/automation/workflows/models.py
index af1d9aadfa..30fa9f6f78 100644
--- a/backend/src/baserow/contrib/automation/workflows/models.py
+++ b/backend/src/baserow/contrib/automation/workflows/models.py
@@ -19,6 +19,7 @@
     OrderableMixin,
     TrashableModelMixin,
 )
+from baserow.core.subjects import UserSubjectType
 
 if TYPE_CHECKING:
     from baserow.contrib.automation.models import Automation
@@ -80,6 +81,19 @@ class AutomationWorkflow(
 
     allow_test_run_until = models.DateTimeField(null=True, blank=True)
 
+    # Who asked for the pending test run or simulation, so a run that waits for
+    # its trigger's event still records them. Cleared with the temporary states.
+    test_run_triggered_by_id = models.PositiveIntegerField(
+        null=True,
+        blank=True,
+        help_text="The id of the subject who asked for the pending test run.",
+    )
+    test_run_triggered_by_type = models.CharField(
+        max_length=255,
+        db_default=UserSubjectType.type,
+        help_text="The subject type of who asked for the pending test run.",
+    )
+
     notification_recipients = models.ManyToManyField(
         settings.AUTH_USER_MODEL,
         blank=True,
diff --git a/backend/src/baserow/contrib/automation/workflows/service.py b/backend/src/baserow/contrib/automation/workflows/service.py
index 8ab5aef301..162c309d0a 100644
--- a/backend/src/baserow/contrib/automation/workflows/service.py
+++ b/backend/src/baserow/contrib/automation/workflows/service.py
@@ -200,7 +200,9 @@ def update_workflow(
             workflow.automation.workspace, kwargs
         )
 
-        updated_workflow = self.handler.update_workflow(workflow, **kwargs)
+        updated_workflow = self.handler.update_workflow(
+            workflow, triggered_by=user, **kwargs
+        )
         automation_workflow_updated.send(
             self, user=user, workflow=updated_workflow.workflow
         )
@@ -354,4 +356,6 @@ def toggle_test_run(
             context=workflow,
         )
 
-        self.handler.toggle_test_run(workflow, simulate_until_node=simulate_until_node)
+        self.handler.toggle_test_run(
+            workflow, simulate_until_node=simulate_until_node, triggered_by=user
+        )
diff --git a/backend/src/baserow/contrib/database/apps.py b/backend/src/baserow/contrib/database/apps.py
index 08e40ede7d..f1448c73b9 100755
--- a/backend/src/baserow/contrib/database/apps.py
+++ b/backend/src/baserow/contrib/database/apps.py
@@ -61,6 +61,10 @@ def ready(self):
 
         action_type_registry.register(ExportTableActionType())
 
+        from .workflow_actions.actions import DispatchButtonFieldActionType
+
+        action_type_registry.register(DispatchButtonFieldActionType())
+
         from .airtable.actions import ImportDatabaseFromAirtableActionType
 
         action_type_registry.register(ImportDatabaseFromAirtableActionType())
diff --git a/backend/src/baserow/contrib/database/rows/handler.py b/backend/src/baserow/contrib/database/rows/handler.py
index a9c7d3bf36..1a727b758c 100644
--- a/backend/src/baserow/contrib/database/rows/handler.py
+++ b/backend/src/baserow/contrib/database/rows/handler.py
@@ -912,127 +912,15 @@ def force_create_row(
         else:
             prepared_values = values
 
-        before_return = before_rows_create.send(
-            self, user=user, table=table, model=model
-        )
-
-        row_values, manytomany_values = self.extract_manytomany_values(
-            prepared_values, model
-        )
-        row_values["order"] = self.get_unique_orders_before_row(before, model)[0]
-
-        if getattr(model, CREATED_BY_COLUMN_NAME, None):
-            row_values[CREATED_BY_COLUMN_NAME] = user if user and user.id else None
-
-        if getattr(model, LAST_MODIFIED_BY_COLUMN_NAME, None):
-            row_values[LAST_MODIFIED_BY_COLUMN_NAME] = (
-                user if user and user.id else None
-            )
-
-        field_rules_handler = FieldRuleHandler(table, user)
-
-        field_rules_handler.on_rows_create([row_values])
-        instance = model(**row_values)
-        field_rules_handler.validate_row(instance)
-
-        def safe_save_instance():
-            try:
-                with transaction.atomic():
-                    instance.save(force_insert=True)
-                rows_created_counter.add(1)
-            except Exception as exc:
-                if is_unique_violation_error(exc):
-                    raise FieldDataConstraintException()
-                else:
-                    raise exc
-
-        try:
-            safe_save_instance()
-        except Exception as exc:
-            if is_index_row_size_error(exc):
-                from baserow.contrib.database.views.handler import (
-                    ViewIndexingHandler,
-                )
-
-                ViewIndexingHandler.handle_index_row_size_error(model.baserow_table_id)
-                safe_save_instance()
-            else:
-                raise exc
-
-        m2m_change_tracker = RowM2MChangeTracker()
-        for field_name, value in manytomany_values.items():
-            m2m_objects, _ = self._prepare_m2m_field_related_objects(
-                instance, field_name, value
-            )
-            field_object = model.get_field_object(field_name)
-            m2m_change_tracker.track_m2m_created_for_new_row(
-                instance,
-                field_object["field"],
-                value,
-            )
-            getattr(instance, field_name).through.objects.bulk_create(m2m_objects)
-
-        cascade_update = field_rules_handler.collector.get_processed_rows()
-
-        fields, dependant_fields, dependant_rows_updates = (
-            self.update_dependencies_of_rows_created(model, [instance])
-        )
-
-        _, cascade_dependant_rows_updates = self.update_dependencies_of_rows_updated(
-            table=table,
-            model=model,
-            updated_rows=cascade_update.updated_rows,
-            updated_field_ids=cascade_update.field_ids,
-        )
-        dependant_rows_updates = merge_dependant_rows_updates(
-            dependant_rows_updates,
-            cascade_dependant_rows_updates,
-            self.cascade_dependant_rows_update(
-                table,
-                cascade_update.row_ids,
-                cascade_update.field_ids,
-                exclude_row_ids=[instance.id],
-            ),
-        )
-
-        if model.fields_requiring_refresh_after_insert():
-            instance.refresh_from_db(
-                fields=model.fields_requiring_refresh_after_insert()
-            )
-
-        from baserow.contrib.database.views.handler import ViewHandler
-
-        ViewHandler().field_value_updated(fields + dependant_fields)
-        SearchHandler.schedule_update_search_data(
-            table, row_ids=[instance.id] + cascade_update.row_ids
-        )
-
-        if cascade_update.row_ids:
-            updated_rows = list(
-                model.objects.all()
-                .enhance_by_fields()
-                .filter(id__in=list(cascade_update.row_ids))
-            )
-            cascade_update.updated_rows = updated_rows
-
-        rows_created.send(
-            self,
-            rows=[instance],
-            before=before,
-            user=user,
-            table=table,
+        return self.force_create_rows(
+            user,
+            table,
+            [prepared_values],
+            before_row=before,
             model=model,
-            send_realtime_update=True,
             send_webhook_events=send_webhook_events,
-            rows_values_refreshed_from_db=False,
-            m2m_change_tracker=m2m_change_tracker,
-            fields=fields,
-            dependant_fields=dependant_fields,
-            before_return=before_return,
-        )
-        self.send_dependant_rows_updated(user, table, dependant_rows_updates)
-
-        return instance
+            values_already_prepared=True,
+        ).created_rows[0]
 
     # noinspection PyMethodMayBeStatic
     def map_user_field_name_dict_to_internal(
@@ -1154,132 +1042,24 @@ def update_row(
         if model is None:
             model = table.get_model()
 
-        updated_fields_by_name = {}
-        updated_fields = []
-        updated_field_ids = set()
-        for field_id, field in model._field_objects.items():
-            if field_id in values or field["name"] in values:
-                updated_field_ids.add(field_id)
-                updated_fields_by_name[field["name"]] = field["field"]
-                updated_fields.append(field["field"])
-
         self._raise_if_values_contain_hidden_fields(user, view, [values])
         self._check_write_fields_values_permissions(user, model, [values])
 
-        rows = [row]
-        before_return = before_rows_update.send(
-            self,
-            rows=rows,
-            user=user,
-            table=table,
-            model=model,
-            updated_field_ids=updated_field_ids,
-        )
-
         if not values_already_prepared:
-            prepared_values = self.prepare_values(model._field_objects, values)
-        else:
-            prepared_values = values
-
-        row_values, manytomany_values = self.extract_manytomany_values(
-            prepared_values, model
-        )
-        update_row_fields = []
-        for name, value in row_values.items():
-            setattr(row, name, value)
-            update_row_fields.append(name)
+            values = self.prepare_values(model._field_objects, values)
 
-        # This update can remove link row connections with other rows. We need to keep
-        # track of these so we can later update any dependant cells in those rows that
-        # we used to link to. This is a dictionary where the key is the id link row
-        # field in this table, and the value is a set of row ids that this row used to
-        # link to via that link row field.
-        m2m_change_tracker = RowM2MChangeTracker()
-
-        for name, value in manytomany_values.items():
-            field = updated_fields_by_name[name]
-            value = [v if not hasattr(v, "id") else v.id for v in value]
-            m2m_change_tracker.track_m2m_update_for_field_and_row(
-                field, name, row, value
-            )
-            getattr(row, name).set(value)
-
-        field_objects_to_always_update = model.get_field_objects_to_always_update()
-        always_updated_fields = ["updated_on"] + [
-            fo["field"].db_column for fo in field_objects_to_always_update
-        ]
-        for field_object in field_objects_to_always_update:
-            updated_field_ids.add(field_object["field"].id)
-        if getattr(model, LAST_MODIFIED_BY_COLUMN_NAME, None):
-            setattr(row, LAST_MODIFIED_BY_COLUMN_NAME, user if user.id else None)
-            always_updated_fields.append(LAST_MODIFIED_BY_COLUMN_NAME)
-
-        def safe_save_row():
-            try:
-                with transaction.atomic():
-                    row.save(update_fields=update_row_fields + always_updated_fields)
-            except Exception as exc:
-                if is_unique_violation_error(exc):
-                    raise FieldDataConstraintException()
-                else:
-                    raise exc
-
-        try:
-            safe_save_row()
-        except Exception as exc:
-            if is_index_row_size_error(exc):
-                from baserow.contrib.database.views.handler import (
-                    ViewIndexingHandler,
-                )
-
-                ViewIndexingHandler.handle_index_row_size_error(model.baserow_table_id)
-                safe_save_row()
-            else:
-                raise exc
-        rows_updated_counter.add(1)
-
-        dependant_fields, dependant_rows_updates = (
-            self.update_dependencies_of_rows_updated(
-                table, [row], model, updated_field_ids, m2m_change_tracker
-            )
-        )
-
-        updated_field_ids.update(
-            field.id for field in dependant_fields if field.table_id == table.id
-        )
-
-        # We need to refresh here as ExpressionFields might have had their values
-        # updated. Django does not support UPDATE .... RETURNING and so we need to
-        # query for the rows updated values instead.
-        row.refresh_from_db(fields=model.fields_requiring_refresh_after_update())
-
-        from baserow.contrib.database.views.handler import ViewHandler
-
-        ViewHandler().field_value_updated(updated_fields + dependant_fields)
-        SearchHandler.schedule_update_search_data(
+        self.force_update_rows(
+            user,
             table,
-            fields=[f for f in updated_fields if f.id in updated_field_ids],
-            row_ids=[row.id],
-        )
-
-        # rows_before_update is serialized in full so the frontend can
-        # reconstruct the pre-update state for filter/sort/search transitions.
-        # rows can be partial because the frontend merges it onto the existing row.
-        rows_updated.send(
-            self,
-            rows=rows,
-            user=user,
-            table=table,
+            [{**values, "id": row.id}],
             model=model,
-            before_return=before_return,
-            updated_field_ids=updated_field_ids,
-            serialize_only_updated_fields=True,
-            m2m_change_tracker=m2m_change_tracker,
-            fields=[f for f in updated_fields if f.id in updated_field_ids],
-            dependant_fields=dependant_fields,
+            rows_to_update=cast(RowsForUpdate, [row]),
+            values_already_prepared=True,
         )
-        self.send_dependant_rows_updated(user, table, dependant_rows_updates)
-
+        # Preserve the single-row API's in-place update contract, including refreshed
+        # formula values and invalidation of any prefetched relations.
+        row.refresh_from_db()
+        del row._m2m_values
         return row
 
     def send_dependant_rows_updated(
@@ -1428,6 +1208,7 @@ def force_create_rows(
         generate_error_report: bool = False,
         skip_search_update: bool = False,
         signal_params: Optional[Dict] = None,
+        values_already_prepared: bool = False,
     ) -> CreatedRowsData:
         """
         Creates new rows for a given table without checking permissions. It also calls
@@ -1449,6 +1230,8 @@ def force_create_rows(
             cells update later on after many create_rows calls then set this to True
             but make sure you trigger it eventually.
         :param signal_params: Additional parameters that are added to the signal.
+        :param values_already_prepared: Whether to use the supplied values directly,
+            skipping defaults and value preparation.
         :return: The created row instances.
 
         """
@@ -1466,15 +1249,18 @@ def force_create_rows(
         )
 
         report = {}
-        rows_values_with_defaults = self.prepare_values_with_defaults(
-            model._field_objects, rows_values
-        )
-        prepared_rows_values, errors = self.prepare_rows_in_bulk(
-            model._field_objects,
-            rows_values_with_defaults,
-            generate_error_report=generate_error_report,
-        )
-        report.update({index: err for index, err in errors.items()})
+        if values_already_prepared:
+            prepared_rows_values = [values.copy() for values in rows_values]
+        else:
+            rows_values_with_defaults = self.prepare_values_with_defaults(
+                model._field_objects, rows_values
+            )
+            prepared_rows_values, errors = self.prepare_rows_in_bulk(
+                model._field_objects,
+                rows_values_with_defaults,
+                generate_error_report=generate_error_report,
+            )
+            report.update({index: err for index, err in errors.items()})
 
         before_return = before_rows_create.send(
             self, user=user, table=table, model=model
@@ -1579,6 +1365,21 @@ def safe_bulk_create():
             )
         )
 
+        if cascade_updated.updated_rows:
+            _, cascade_dependant_rows_updates = (
+                self.update_dependencies_of_rows_updated(
+                    table=table,
+                    model=model,
+                    updated_rows=cascade_updated.updated_rows,
+                    updated_field_ids=cascade_updated.field_ids,
+                    skip_search_updates=skip_search_update,
+                    collect_dependant_rows=send_realtime_update,
+                )
+            )
+            dependant_rows_updates = merge_dependant_rows_updates(
+                dependant_rows_updates, cascade_dependant_rows_updates
+            )
+
         from baserow.contrib.database.views.handler import ViewHandler
 
         updated_fields = [o["field"] for o in model._field_objects.values()]
@@ -2483,6 +2284,7 @@ def force_update_rows(
         skip_search_update: bool = False,
         generate_error_report: bool = False,
         signal_params: Optional[Dict] = None,
+        values_already_prepared: bool = False,
     ) -> UpdatedRowsData:
         """
         Updates field values in batch based on provided rows with the new
@@ -2505,6 +2307,8 @@ def force_update_rows(
             but make sure you trigger it eventually.
         :param generate_error_report: Generate error report if set to True.
         :param signal_params: Additional parameters that are added to the signal.
+        :param values_already_prepared: Whether the values are already sanitized and
+            validated and can be used without preparing them again.
         :raises RowIdsNotUnique: When trying to update the same row multiple
             times.
         :raises RowDoesNotExist: When any of the rows don't exist.
@@ -2521,12 +2325,17 @@ def force_update_rows(
 
         user_id = user and user.id
 
-        prepared_rows_values, errors = self.prepare_rows_in_bulk(
-            model._field_objects,
-            rows_values,
-            generate_error_report=generate_error_report,
-        )
-        report = {index: err for index, err in errors.items()}
+        if values_already_prepared:
+            # Row IDs are removed below, so do not mutate the caller's dictionaries.
+            prepared_rows_values = [values.copy() for values in rows_values]
+            report = {}
+        else:
+            prepared_rows_values, errors = self.prepare_rows_in_bulk(
+                model._field_objects,
+                rows_values,
+                generate_error_report=generate_error_report,
+            )
+            report = {index: err for index, err in errors.items()}
         row_ids = [r["id"] for r in prepared_rows_values]
 
         non_unique_ids = get_non_unique_values(row_ids)
@@ -2649,15 +2458,15 @@ def force_update_rows(
                 # rows which previously were connected to an updated row, but no
                 # longer are.
                 field_obj = field_name_to_field[field_name]
+                # Prepared values can contain model instances. Track and compare IDs.
+                value_ids = [v.id if hasattr(v, "id") else v for v in value]
                 m2m_change_tracker.track_m2m_update_for_field_and_row(
-                    field_obj, field_name, row, value
+                    field_obj, field_name, row, value_ids
                 )
 
                 original_set_of_values = set(
                     original_row_values_by_id[row.id].get(field_name) or []
                 )
-                # if a list of models is provided as value, make sure to compare the ids
-                value_ids = [v.id if hasattr(v, "id") else v for v in value]
                 new_set_of_values = set(value_ids)
                 to_add = new_set_of_values - original_set_of_values
                 to_delete = original_set_of_values - new_set_of_values
@@ -2695,7 +2504,9 @@ def force_update_rows(
         for field_name, m2m_to_add in m2m_values_to_add.items():
             through = getattr(model, field_name).through
             row_column_name = row_column_names[field_name]
-            through.objects.bulk_create(m2m_to_add)
+            # Related managers hide trashed rows even though their through rows
+            # still exist. Match Django's set() when inserting those links again.
+            through.objects.bulk_create(m2m_to_add, ignore_conflicts=True)
 
         bulk_update_fields = ["updated_on"]
         if field_rules_handler.has_field_rules():
diff --git a/backend/src/baserow/contrib/database/workflow_actions/actions.py b/backend/src/baserow/contrib/database/workflow_actions/actions.py
new file mode 100644
index 0000000000..9ea873b0be
--- /dev/null
+++ b/backend/src/baserow/contrib/database/workflow_actions/actions.py
@@ -0,0 +1,79 @@
+import dataclasses
+from typing import Any
+
+from django.contrib.auth.models import AbstractUser
+from django.utils.translation import gettext_lazy as _
+
+from baserow.contrib.database.action.scopes import (
+    TABLE_ACTION_CONTEXT,
+    TableActionScopeType,
+)
+from baserow.contrib.database.fields.models import ButtonField
+from baserow.core.action.registries import ActionType, ActionTypeDescription
+
+
+class DispatchButtonFieldActionType(ActionType):
+    """A button click, recorded for the audit log. Not undoable (ADR 006 s.8)."""
+
+    type = "dispatch_button_field"
+    description = ActionTypeDescription(
+        _("Click button"),
+        _('Button "%(field_name)s" (%(field_id)s) clicked on row %(row_id)s'),
+        TABLE_ACTION_CONTEXT,
+    )
+    analytics_params = [
+        "table_id",
+        "database_id",
+        "workspace_id",
+        "field_id",
+        "action_count",
+    ]
+
+    @dataclasses.dataclass
+    class Params:
+        table_id: int
+        table_name: str
+        database_id: int
+        database_name: str
+        workspace_id: int
+        workspace_name: str
+        field_id: int
+        field_name: str
+        row_id: int
+        action_count: int
+
+    @classmethod
+    def do(cls, user: AbstractUser, field: ButtonField, row: Any, action_count: int):
+        """
+        Records the click.
+
+        :param user: The clicker.
+        :param field: The clicked button field.
+        :param row: The clicked row, a generated table model instance.
+        :param action_count: How many actions the button carried at the click.
+        """
+
+        table = field.table
+        database = table.database
+        workspace = database.workspace
+        cls.register_action(
+            user,
+            cls.Params(
+                table.id,
+                table.name,
+                database.id,
+                database.name,
+                workspace.id,
+                workspace.name,
+                field.id,
+                field.name,
+                row.id,
+                action_count,
+            ),
+            cls.scope(table.id),
+            workspace,
+        )
+
+    @classmethod
+    def scope(cls, table_id: int):
+        return TableActionScopeType.value(table_id)
diff --git a/backend/src/baserow/contrib/database/workflow_actions/dispatch_context.py b/backend/src/baserow/contrib/database/workflow_actions/dispatch_context.py
index 54da406833..a3c1cc5926 100644
--- a/backend/src/baserow/contrib/database/workflow_actions/dispatch_context.py
+++ b/backend/src/baserow/contrib/database/workflow_actions/dispatch_context.py
@@ -27,7 +27,7 @@ class DatabaseDispatchContext(DispatchContext):
     context, and nothing dispatched by a click is a paginated list service.
     """
 
-    own_properties = ["field", "row"]
+    own_properties = ["field", "row", "actor"]
 
     def __init__(
         self,
@@ -43,17 +43,15 @@ def __init__(
         :param row: The clicked row, as a generated table model instance.
         """
 
-        # Everything defaults to None only so `clone()` can rebuild this class
-        # without passing `actor`. Neither `field` nor `row` is optional for a
-        # real dispatch, so fail here rather than inside a data provider later.
+        # `field` and `row` default to None only because they follow `actor`.
+        # Neither is optional for a real dispatch, so fail here rather than
+        # inside a data provider later.
         if field is None or row is None:
             raise TypeError("DatabaseDispatchContext requires field and row")
 
         self.field = field
         self.row = row
 
-        # `clone()` carries `actor` over itself, hence its absence from
-        # `own_properties`.
         super().__init__(actor=actor, **kwargs)
 
         # Holds the row read for the action that is running. It has to be made
diff --git a/backend/src/baserow/contrib/database/workflow_actions/service.py b/backend/src/baserow/contrib/database/workflow_actions/service.py
index d6ca145bf0..15a2cf9cc5 100644
--- a/backend/src/baserow/contrib/database/workflow_actions/service.py
+++ b/backend/src/baserow/contrib/database/workflow_actions/service.py
@@ -15,6 +15,9 @@
     ReadFieldOperationType,
     UpdateFieldOperationType,
 )
+from baserow.contrib.database.workflow_actions.actions import (
+    DispatchButtonFieldActionType,
+)
 from baserow.contrib.database.workflow_actions.dispatch_context import (
     DatabaseDispatchContext,
 )
@@ -579,6 +582,7 @@ def dispatch_workflow_actions(
         # Nothing server side means no state to protect, so no lock: a button
         # that only opens a URL must not reject a second click.
         if not server_actions:
+            DispatchButtonFieldActionType.do(user, field, row, len(workflow_actions))
             return WorkflowActionsDispatchResult(
                 client_actions=client_actions, positions=positions
             )
@@ -606,6 +610,9 @@ def dispatch_workflow_actions(
             raise WorkflowActionDispatchInProgress()
 
         try:
+            # Inside the lock, so a click refused as already running leaves no entry.
+            DispatchButtonFieldActionType.do(user, field, row, len(workflow_actions))
+
             # Remembering a result edits the button's configuration, so it
             # follows the field's update permission rather than the lower bar
             # for clicking (ADR 006 section 7). Only asked when an action of
diff --git a/backend/src/baserow/contrib/integrations/core/service_types.py b/backend/src/baserow/contrib/integrations/core/service_types.py
index 8fc03232aa..2d9bf3b9c2 100644
--- a/backend/src/baserow/contrib/integrations/core/service_types.py
+++ b/backend/src/baserow/contrib/integrations/core/service_types.py
@@ -2517,7 +2517,7 @@ class CoreStartWorkflowServiceType(CoreServiceType):
     WORKFLOW_DOES_NOT_EXIST_ERROR = "The workflow with ID {workflow_id} does not exist."
     TRIGGER_NOT_ON_DEMAND_ERROR = (
         "Only workflows whose trigger can start on demand, such as a manual "
-        "trigger, can be started."
+        "or periodic trigger, can be started."
     )
 
     allowed_fields = ["workflow"]
@@ -2621,7 +2621,11 @@ def dispatch_data(
                 self.TRIGGER_NOT_ON_DEMAND_ERROR
             )
 
-        AutomationWorkflowHandler().async_start_workflow(published_workflow)
+        # Only a button click has an actor here; a nested start-workflow node
+        # and the builder action do not, so their runs record nobody.
+        AutomationWorkflowHandler().async_start_workflow(
+            published_workflow, triggered_by=dispatch_context.actor
+        )
         return None
 
     def dispatch_transform(
diff --git a/backend/src/baserow/core/registries.py b/backend/src/baserow/core/registries.py
index a4d6bf8534..16192d3170 100755
--- a/backend/src/baserow/core/registries.py
+++ b/backend/src/baserow/core/registries.py
@@ -1292,6 +1292,21 @@ def get_serializer(self, model_instance, **kwargs) -> Serializer:
         instance_type = self.get_by_model(model_instance)
         return instance_type.get_serializer(model_instance, **kwargs)
 
+    def get_subject(self, type_name: str, subject_id: int) -> Optional[Subject]:
+        """
+        Returns the subject stored as a type and id pair.
+
+        :param type_name: The subject type, for example `auth.User`.
+        :param subject_id: The id of the subject.
+        :return: The subject, or None when it no longer exists or its type is not
+            stored in the database.
+        """
+
+        manager = getattr(self.get(type_name).model_class, "objects", None)
+        if manager is None:
+            return None
+        return manager.filter(id=subject_id).first()
+
 
 class OperationType(abc.ABC, Instance):
     """
diff --git a/backend/src/baserow/core/services/dispatch_context.py b/backend/src/baserow/core/services/dispatch_context.py
index 505c1e9343..f0260032da 100644
--- a/backend/src/baserow/core/services/dispatch_context.py
+++ b/backend/src/baserow/core/services/dispatch_context.py
@@ -16,6 +16,7 @@ class DispatchContext(RuntimeFormulaContext, ABC):
         "use_sample_data",
         "force_outputs",
         "event_payload",
+        "actor",
     ]
 
     """
@@ -76,9 +77,6 @@ def clone(self, **kwargs) -> RuntimeFormulaContextSubClass:
         """
         Return a new DispatchContext instance cloned from the current context, without
         losing the original cached data and call stack but updating some properties.
-
-        The actor is carried over explicitly rather than via `own_properties`, so
-        it survives subclasses that replace that list.
         """
 
         new_values = {}
@@ -86,9 +84,7 @@ def clone(self, **kwargs) -> RuntimeFormulaContextSubClass:
             new_values[prop] = getattr(self, prop)
         new_values.update(kwargs)
 
-        actor = new_values.pop("actor", self.actor)
         new_context = self.__class__(**new_values)
-        new_context.actor = actor
         new_context.cache = {**self.cache}
         new_context.call_stack = set(self.call_stack)
 
diff --git a/backend/src/baserow/test_utils/fixtures/automation_history.py b/backend/src/baserow/test_utils/fixtures/automation_history.py
index f2930e32e9..327c3a1703 100644
--- a/backend/src/baserow/test_utils/fixtures/automation_history.py
+++ b/backend/src/baserow/test_utils/fixtures/automation_history.py
@@ -35,17 +35,20 @@ def create_workflow_history(self, user=None, **kwargs):
         if status is None:
             status = HistoryStatusChoices.SUCCESS
 
-        is_test_run = kwargs.pop("status", False)
+        is_test_run = kwargs.pop("is_test_run", False)
 
         self.create_local_baserow_create_row_action_node(
             user=user, workflow=original_workflow
         )
 
+        # Anything left is the handler's to accept or refuse, so a misspelt
+        # name fails here instead of quietly building a default row.
         history = AutomationHistoryHandler().create_workflow_history(
             original_workflow=original_workflow,
             workflow=original_workflow,
             started_on=started_on,
             is_test_run=is_test_run,
+            **kwargs,
         )
 
         history.completed_on = completed_on
diff --git a/backend/src/baserow/test_utils/pytest_conftest.py b/backend/src/baserow/test_utils/pytest_conftest.py
index d17b69d47f..50f8fd9251 100755
--- a/backend/src/baserow/test_utils/pytest_conftest.py
+++ b/backend/src/baserow/test_utils/pytest_conftest.py
@@ -991,6 +991,7 @@ class FakeDispatchContext(DispatchContext):
         "_is_publicly_sortable",
         "_filters",
         "_sortings",
+        "actor",
     ]
 
     def __init__(self, **kwargs):
diff --git a/backend/tests/baserow/contrib/automation/api/nodes/test_nodes_views.py b/backend/tests/baserow/contrib/automation/api/nodes/test_nodes_views.py
index f734d68c9c..b9f754fa8c 100644
--- a/backend/tests/baserow/contrib/automation/api/nodes/test_nodes_views.py
+++ b/backend/tests/baserow/contrib/automation/api/nodes/test_nodes_views.py
@@ -1005,7 +1005,7 @@ def test_simulate_dispatch_trigger_node_immediate_dispatch(
 
     assert workflow.simulate_until_node_id == trigger_node.id
     # In case of an immediate dispatch we want to trigger immediately the workflow
-    mock_async_start_workflow.assert_called_with(workflow)
+    mock_async_start_workflow.assert_called_with(workflow, triggered_by=user)
 
     trigger_node.service.get_type().can_be_immediately_dispatched = old_imm
 
@@ -1166,7 +1166,7 @@ def test_simulate_dispatch_action_node_with_sample_data(
     assert response.status_code == HTTP_202_ACCEPTED
 
     # As the trigger node has sample data we can immediately trigger the workflow
-    mock_async_start_workflow.assert_called_with(workflow)
+    mock_async_start_workflow.assert_called_with(workflow, triggered_by=user)
 
     workflow.refresh_from_db()
     assert workflow.simulate_until_node_id == action_node.id
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 27d9bff843..4add8a1508 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
@@ -1,4 +1,5 @@
 import datetime
+from unittest.mock import patch
 
 from django.db.models import Count, Q
 from django.urls import reverse
@@ -20,7 +21,11 @@
 )
 from baserow.contrib.automation.history.constants import HistoryStatusChoices
 from baserow.contrib.automation.history.handler import AutomationHistoryHandler
-from baserow.contrib.automation.nodes.node_types import CorePeriodicTriggerNodeType
+from baserow.contrib.automation.history.models import AutomationWorkflowHistory
+from baserow.contrib.automation.nodes.node_types import (
+    CoreManualTriggerNodeType,
+    CorePeriodicTriggerNodeType,
+)
 from baserow.contrib.automation.workflows.constants import ALLOW_TEST_RUN_MINUTES
 from baserow.contrib.database.rows.handler import RowHandler
 from baserow.core.cache import local_cache
@@ -478,6 +483,27 @@ def test_enable_workflow_test_run(api_client, data_fixture):
     )
 
 
+@pytest.mark.django_db
+def test_a_test_run_records_who_started_it(api_client, data_fixture):
+    user, token = data_fixture.create_user_and_token()
+    workflow = data_fixture.create_automation_workflow(
+        user, trigger_type=CoreManualTriggerNodeType.type
+    )
+    url = reverse(API_URL_WORKFLOW_TEST, kwargs={"workflow_id": workflow.id})
+
+    with patch(
+        "baserow.contrib.automation.workflows.handler.start_workflow_celery_task"
+    ):
+        response = api_client.post(
+            url, format="json", HTTP_AUTHORIZATION=f"JWT {token}"
+        )
+
+    assert response.status_code == HTTP_202_ACCEPTED
+    history = AutomationWorkflowHistory.objects.get(original_workflow=workflow)
+    assert history.is_test_run is True
+    assert history.triggered_by_id == user.id
+
+
 @pytest.mark.django_db
 def test_disable_workflow_test_run(api_client, data_fixture):
     user, token = data_fixture.create_user_and_token()
@@ -681,11 +707,28 @@ def test_get_workflow_histories(api_client, data_fixture):
                 "status": "success",
                 "simulate_until_node": None,
                 "plugin_data": {},
+                "triggered_by": None,
             },
         ],
     }
 
 
+@pytest.mark.django_db
+def test_get_workflow_histories_names_who_triggered_the_run(api_client, data_fixture):
+    user, token = data_fixture.create_user_and_token(first_name="Ada")
+    history = data_fixture.create_workflow_history(user=user, triggered_by=user)
+
+    url = reverse(API_URL_WORKFLOW_HISTORY, kwargs={"workflow_id": history.workflow.id})
+    response = api_client.get(url, **get_api_kwargs(token))
+
+    assert response.status_code == HTTP_200_OK
+    assert response.json()["results"][0]["triggered_by"] == {
+        "id": user.id,
+        "type": "auth.User",
+        "name": "Ada",
+    }
+
+
 @pytest.mark.django_db
 def test_get_workflow_histories_invalid_workflow(api_client, data_fixture):
     user, token = data_fixture.create_user_and_token()
@@ -753,12 +796,15 @@ def test_get_workflow_histories_query_count(data_fixture, django_assert_num_quer
     handler = AutomationHistoryHandler()
 
     def _create_histories(count):
-        for _ in range(count):
+        for index in range(count):
             workflow_history = handler.create_workflow_history(
                 original_workflow=workflow,
                 workflow=workflow,
                 started_on=timezone.now(),
                 is_test_run=False,
+                # Every other run names its user, so the serializer's
+                # nested starter is exercised without a query per row.
+                triggered_by=user if index % 2 else None,
             )
             node_history = handler.create_node_history(
                 workflow_history=workflow_history,
diff --git a/backend/tests/baserow/contrib/automation/history/test_history_handler.py b/backend/tests/baserow/contrib/automation/history/test_history_handler.py
index f366dd5af7..6b51e235bf 100644
--- a/backend/tests/baserow/contrib/automation/history/test_history_handler.py
+++ b/backend/tests/baserow/contrib/automation/history/test_history_handler.py
@@ -559,3 +559,57 @@ def test_get_edge_labels_returns_expected_data(data_fixture):
         node_history_1.id: "foo label",
         node_history_2.id: "bar label",
     }
+
+
+@pytest.mark.django_db
+def test_create_workflow_history_records_who_triggered_it(data_fixture):
+    user = data_fixture.create_user()
+    workflow = data_fixture.create_automation_workflow(user=user)
+
+    history = AutomationHistoryHandler().create_workflow_history(
+        original_workflow=workflow,
+        workflow=workflow,
+        started_on=timezone.now(),
+        is_test_run=False,
+        triggered_by=user,
+    )
+
+    history.refresh_from_db()
+    assert history.triggered_by_id == user.id
+    assert history.triggered_by_type == "auth.User"
+    assert history.triggered_by_name == user.first_name
+
+
+@pytest.mark.django_db
+def test_create_workflow_history_defaults_to_nobody(data_fixture):
+    workflow = data_fixture.create_automation_workflow()
+
+    history = AutomationHistoryHandler().create_workflow_history(
+        original_workflow=workflow,
+        workflow=workflow,
+        started_on=timezone.now(),
+        is_test_run=False,
+    )
+
+    assert history.triggered_by_id is None
+    assert history.triggered_by_name == ""
+
+
+@pytest.mark.django_db
+def test_deleting_the_user_keeps_who_started_the_run(data_fixture):
+    user = data_fixture.create_user(first_name="Ada")
+    workflow = data_fixture.create_automation_workflow()
+
+    history = AutomationHistoryHandler().create_workflow_history(
+        original_workflow=workflow,
+        workflow=workflow,
+        started_on=timezone.now(),
+        is_test_run=False,
+        triggered_by=user,
+    )
+    user_id = user.id
+    user.delete()
+
+    history.refresh_from_db()
+    assert history.triggered_by_id == user_id
+    assert history.triggered_by_name == "Ada"
diff --git a/backend/tests/baserow/contrib/automation/nodes/test_node_dispatch_trigger_user.py b/backend/tests/baserow/contrib/automation/nodes/test_node_dispatch_trigger_user.py
new file mode 100644
index 0000000000..8f7bdb7292
--- /dev/null
+++ b/backend/tests/baserow/contrib/automation/nodes/test_node_dispatch_trigger_user.py
@@ -0,0 +1,37 @@
+from unittest.mock import patch
+
+import pytest
+
+from baserow.contrib.automation.nodes.handler import AutomationNodeHandler
+from baserow.contrib.automation.nodes.node_types import CoreManualTriggerNodeType
+from baserow.core.services.types import DispatchResult
+
+
+def _capture_context(captured):
+    """A node type `dispatch` that records the context it was handed."""
+
+    def dispatch(self, node, dispatch_context):
+        captured.append(dispatch_context)
+        return DispatchResult(data={})
+
+    return dispatch
+
+
+@pytest.mark.django_db
+def test_dispatch_node_does_not_act_as_who_started_the_run(data_fixture):
+    user = data_fixture.create_user()
+    workflow = data_fixture.create_automation_workflow(
+        user=user, trigger_type=CoreManualTriggerNodeType.type
+    )
+    trigger = workflow.get_trigger()
+    history = data_fixture.create_automation_workflow_history(
+        workflow=workflow, triggered_by_id=user.id, triggered_by_type="auth.User"
+    )
+    captured = []
+
+    with patch.object(
+        CoreManualTriggerNodeType, "dispatch", _capture_context(captured)
+    ):
+        AutomationNodeHandler().dispatch_node(trigger.id, history.id)
+
+    assert [context.actor for context in captured] == [None]
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 0e8428d609..e498807197 100644
--- a/backend/tests/baserow/contrib/automation/workflows/test_workflow_handler.py
+++ b/backend/tests/baserow/contrib/automation/workflows/test_workflow_handler.py
@@ -20,6 +20,9 @@
     CorePeriodicTriggerNodeType,
     LocalBaserowRowsCreatedNodeTriggerType,
 )
+from baserow.contrib.automation.workflows.actions import (
+    UpdateAutomationWorkflowActionType,
+)
 from baserow.contrib.automation.workflows.constants import (
     ALLOW_TEST_RUN_MINUTES,
     WORKFLOW_DIRTY_CACHE_KEY,
@@ -33,9 +36,11 @@
     AutomationWorkflowTooManyErrors,
 )
 from baserow.contrib.automation.workflows.handler import AutomationWorkflowHandler
+from baserow.core.action.handler import ActionHandler
 from baserow.core.cache import global_cache, local_cache
 from baserow.core.notifications.models import Notification, NotificationRecipient
 from baserow.core.registries import ImportExportConfig
+from baserow.core.subjects import UserSubjectType
 from baserow.core.trash.handler import TrashHandler
 from tests.baserow.contrib.automation.history.utils import assert_history
 
@@ -1946,3 +1951,232 @@ def test_async_start_workflow_test_run_creates_test_clone(
     mock_start_workflow_celery_task.delay.assert_called_once_with(
         history.workflow_id, history.id
     )
+
+
+@pytest.mark.django_db
+def test_async_start_workflow_records_who_triggered_it(data_fixture):
+    user = data_fixture.create_user()
+    workflow = data_fixture.create_automation_workflow(user=user)
+    published = AutomationWorkflowHandler().publish(workflow)
+
+    with patch(
+        "baserow.contrib.automation.workflows.handler.start_workflow_celery_task"
+    ):
+        AutomationWorkflowHandler().async_start_workflow(published, triggered_by=user)
+
+    history = AutomationWorkflowHistory.objects.get(original_workflow=workflow)
+    assert history.triggered_by_id == user.id
+
+
+@pytest.mark.django_db
+def test_async_start_workflow_records_the_trigger_on_a_refused_run(data_fixture):
+    """The rate limited branch writes its own history row; it names the user too."""
+
+    user = data_fixture.create_user()
+    workflow = data_fixture.create_automation_workflow(user=user)
+    published = AutomationWorkflowHandler().publish(workflow)
+
+    with patch.object(
+        AutomationWorkflowHandler,
+        "before_run",
+        side_effect=AutomationWorkflowRateLimited("too fast"),
+    ):
+        AutomationWorkflowHandler().async_start_workflow(published, triggered_by=user)
+
+    history = AutomationWorkflowHistory.objects.get(original_workflow=workflow)
+    assert history.status == "error"
+    assert history.triggered_by_id == user.id
+
+
+def _fire_rows_created_event(workflow, table):
+    trigger = workflow.get_trigger()
+    service = trigger.service.specific
+    trigger.get_type().on_event(
+        service.get_type().model_class.objects.filter(table=table),
+        [{"id": 1, "order": "1.00000000000000000000"}],
+    )
+
+
+@pytest.mark.django_db
+def test_a_test_run_waiting_for_its_event_records_who_started_it(data_fixture):
+    user = data_fixture.create_user()
+    table = data_fixture.create_database_table(user=user)
+    workflow = data_fixture.create_automation_workflow(
+        user,
+        trigger_type=LocalBaserowRowsCreatedNodeTriggerType.type,
+        trigger_service_kwargs={"table": table},
+    )
+
+    with patch(f"{WORKFLOWS_MODULE}.handler.start_workflow_celery_task"):
+        AutomationWorkflowHandler().toggle_test_run(
+            workflow, simulate_until_node=None, triggered_by=user
+        )
+        assert not AutomationWorkflowHistory.objects.filter(
+            original_workflow=workflow
+        ).exists()
+
+        _fire_rows_created_event(workflow, table)
+
+    history = AutomationWorkflowHistory.objects.get(original_workflow=workflow)
+    assert history.is_test_run is True
+    assert history.triggered_by_id == user.id
+    workflow.refresh_from_db()
+    assert workflow.test_run_triggered_by_id is None
+
+
+@pytest.mark.django_db
+def test_a_simulation_waiting_for_its_event_records_who_started_it(data_fixture):
+    user = data_fixture.create_user()
+    table = data_fixture.create_database_table(user=user)
+    workflow = data_fixture.create_automation_workflow(
+        user,
+        trigger_type=LocalBaserowRowsCreatedNodeTriggerType.type,
+        trigger_service_kwargs={"table": table},
+    )
+
+    with patch(f"{WORKFLOWS_MODULE}.handler.start_workflow_celery_task"):
+        AutomationWorkflowHandler().toggle_test_run(
+            workflow, simulate_until_node=workflow.get_trigger(), triggered_by=user
+        )
+        assert not AutomationWorkflowHistory.objects.filter(
+            original_workflow=workflow
+        ).exists()
+
+        _fire_rows_created_event(workflow, table)
+
+    history = AutomationWorkflowHistory.objects.get(original_workflow=workflow)
+    assert history.simulate_until_node_id == workflow.get_trigger().id
+    assert history.triggered_by_id == user.id
+    workflow.refresh_from_db()
+    assert workflow.test_run_triggered_by_id is None
+
+
+@pytest.mark.django_db
+def test_cancelling_a_waiting_test_run_forgets_who_started_it(data_fixture):
+    user = data_fixture.create_user()
+    workflow = data_fixture.create_automation_workflow(
+        user, trigger_type=LocalBaserowRowsCreatedNodeTriggerType.type
+    )
+    handler = AutomationWorkflowHandler()
+
+    handler.toggle_test_run(workflow, simulate_until_node=None, triggered_by=user)
+    workflow.refresh_from_db()
+    assert workflow.test_run_triggered_by_id == user.id
+
+    handler.toggle_test_run(workflow, simulate_until_node=None)
+    workflow.refresh_from_db()
+    assert workflow.test_run_triggered_by_id is None
+    assert handler.get_test_run_triggered_by(workflow) is None
+
+
+@pytest.mark.django_db
+def test_a_waiting_test_run_whose_starter_was_deleted_records_nobody(data_fixture):
+    user = data_fixture.create_user()
+    starter = data_fixture.create_user()
+    workflow = data_fixture.create_automation_workflow(
+        user, trigger_type=LocalBaserowRowsCreatedNodeTriggerType.type
+    )
+
+    AutomationWorkflowHandler().toggle_test_run(
+        workflow, simulate_until_node=None, triggered_by=starter
+    )
+    starter.delete()
+    workflow.refresh_from_db()
+
+    assert AutomationWorkflowHandler().get_test_run_triggered_by(workflow) is None
+
+
+@pytest.mark.django_db
+def test_opening_the_test_run_window_through_an_update_records_who_opened_it(
+    data_fixture,
+):
+    first = data_fixture.create_user()
+    workspace = data_fixture.create_workspace(user=first)
+    second = data_fixture.create_user(workspace=workspace)
+    automation = data_fixture.create_automation_application(workspace=workspace)
+    table = data_fixture.create_database_table(user=first)
+    workflow = data_fixture.create_automation_workflow(
+        first,
+        automation=automation,
+        trigger_type=LocalBaserowRowsCreatedNodeTriggerType.type,
+        trigger_service_kwargs={"table": table},
+    )
+
+    AutomationWorkflowHandler().toggle_test_run(
+        workflow, simulate_until_node=None, triggered_by=first
+    )
+    UpdateAutomationWorkflowActionType.do(first, workflow.id, {"allow_test_run": False})
+    workflow.refresh_from_db()
+    assert workflow.test_run_triggered_by_id is None
+
+    UpdateAutomationWorkflowActionType.do(second, workflow.id, {"allow_test_run": True})
+    workflow.refresh_from_db()
+    assert workflow.test_run_triggered_by_id == second.id
+
+    with patch(f"{WORKFLOWS_MODULE}.handler.start_workflow_celery_task"):
+        _fire_rows_created_event(workflow, table)
+
+    history = AutomationWorkflowHistory.objects.get(original_workflow=workflow)
+    assert history.triggered_by_id == second.id
+
+
+@pytest.mark.django_db
+def test_resetting_temporary_states_forgets_the_starter_type(data_fixture):
+    user = data_fixture.create_user()
+    workflow = data_fixture.create_automation_workflow(
+        user, trigger_type=LocalBaserowRowsCreatedNodeTriggerType.type
+    )
+    workflow.test_run_triggered_by_id = 1
+    workflow.test_run_triggered_by_type = "core.Token"
+    workflow.save()
+
+    AutomationWorkflowHandler().reset_workflow_temporary_states(workflow)
+
+    workflow.refresh_from_db()
+    assert workflow.test_run_triggered_by_id is None
+    assert workflow.test_run_triggered_by_type == UserSubjectType.type
+
+
+@pytest.mark.django_db
+def test_scheduling_a_test_run_without_a_starter_forgets_an_earlier_one(
+    data_fixture,
+):
+    user = data_fixture.create_user()
+    workflow = data_fixture.create_automation_workflow(
+        user, trigger_type=LocalBaserowRowsCreatedNodeTriggerType.type
+    )
+    handler = AutomationWorkflowHandler()
+    workflow.test_run_triggered_by_id = user.id
+    workflow.save()
+
+    handler.set_workflow_temporary_states(workflow)
+
+    workflow.refresh_from_db()
+    assert workflow.test_run_triggered_by_id is None
+
+
+@pytest.mark.django_db
+@pytest.mark.undo_redo
+def test_undoing_an_unrelated_update_keeps_the_test_run_starter(data_fixture):
+    first = data_fixture.create_user()
+    workspace = data_fixture.create_workspace(user=first)
+    session_id = "undo-keeps-starter"
+    second = data_fixture.create_user(workspace=workspace, session_id=session_id)
+    automation = data_fixture.create_automation_application(workspace=workspace)
+    workflow = data_fixture.create_automation_workflow(
+        first,
+        automation=automation,
+        trigger_type=LocalBaserowRowsCreatedNodeTriggerType.type,
+    )
+    AutomationWorkflowHandler().toggle_test_run(
+        workflow, simulate_until_node=None, triggered_by=first
+    )
+
+    UpdateAutomationWorkflowActionType.do(second, workflow.id, {"name": "Renamed"})
+    ActionHandler.undo(
+        second, [UpdateAutomationWorkflowActionType.scope(automation.id)], session_id
+    )
+
+    workflow.refresh_from_db()
+    assert workflow.name != "Renamed"
+    assert workflow.test_run_triggered_by_id == first.id
diff --git a/backend/tests/baserow/contrib/dashboard/data_sources/test_dashboard_dispatch_context.py b/backend/tests/baserow/contrib/dashboard/data_sources/test_dashboard_dispatch_context.py
index 27f11bf3ef..9f862e4dc8 100644
--- a/backend/tests/baserow/contrib/dashboard/data_sources/test_dashboard_dispatch_context.py
+++ b/backend/tests/baserow/contrib/dashboard/data_sources/test_dashboard_dispatch_context.py
@@ -20,6 +20,8 @@ def test_dispatch_context_clone(data_fixture):
     assert new_dispatch_context.request is request
     assert new_dispatch_context.widget == widget
     assert new_dispatch_context.cache == {"key": "value"}
+    # Only contexts that take an actor carry it; the rest keep the base None.
+    assert new_dispatch_context.actor is None
 
 
 @pytest.mark.django_db
diff --git a/backend/tests/baserow/contrib/database/field/test_multiple_collaborators_field_type.py b/backend/tests/baserow/contrib/database/field/test_multiple_collaborators_field_type.py
index 939113c86e..9206a7840b 100644
--- a/backend/tests/baserow/contrib/database/field/test_multiple_collaborators_field_type.py
+++ b/backend/tests/baserow/contrib/database/field/test_multiple_collaborators_field_type.py
@@ -1057,11 +1057,12 @@ def export_row(row):
                 getattr(row, field_object["name"]), field_object
             )
 
-    # Let's count the number of queries to export one row
+    # Creation prefetches the collaborators, so only the formula's user-details
+    # query is needed when exporting the first row.
     with CaptureQueriesContext(connection) as queries_for_first:
         export_row(first_row)
 
-    assert len(queries_for_first.captured_queries) == 2
+    assert len(queries_for_first.captured_queries) == 1
 
     other_rows = row_handler.force_create_rows(
         user=user,
diff --git a/backend/tests/baserow/contrib/database/rows/test_rows_actions.py b/backend/tests/baserow/contrib/database/rows/test_rows_actions.py
index 206f2ec20c..c5a8bd3368 100644
--- a/backend/tests/baserow/contrib/database/rows/test_rows_actions.py
+++ b/backend/tests/baserow/contrib/database/rows/test_rows_actions.py
@@ -57,7 +57,7 @@ def test_can_undo_creating_row(data_fixture):
     assert model.objects.all().count() == 1
     assert getattr(row, f"field_{name_field.id}") == "Tesla"
     assert getattr(row, f"field_{speed_field.id}") == 240
-    assert getattr(row, f"field_{price_field.id}") == 59999.99
+    assert getattr(row, f"field_{price_field.id}") == Decimal("59999.99")
     assert not getattr(row, "field_9999", None)
 
     action_undone = ActionHandler.undo(
@@ -103,7 +103,7 @@ def test_can_undo_redo_creating_row(data_fixture):
     assert model.objects.all().count() == 1
     assert getattr(row, f"field_{name_field.id}") == "Tesla"
     assert getattr(row, f"field_{speed_field.id}") == 240
-    assert getattr(row, f"field_{price_field.id}") == 59999.99
+    assert getattr(row, f"field_{price_field.id}") == Decimal("59999.99")
     assert not getattr(row, "field_9999", None)
 
     ActionHandler.undo(
@@ -120,7 +120,7 @@ def test_can_undo_redo_creating_row(data_fixture):
 
     assert getattr(row, f"field_{name_field.id}") == "Tesla"
     assert getattr(row, f"field_{speed_field.id}") == 240
-    assert getattr(row, f"field_{price_field.id}") == 59999.99
+    assert getattr(row, f"field_{price_field.id}") == Decimal("59999.99")
     assert not getattr(row, "field_9999", None)
 
 
@@ -440,7 +440,7 @@ def test_can_undo_deleting_row(data_fixture):
     assert model.objects.all().count() == 1
     assert getattr(row, f"field_{name_field.id}") == "Tesla"
     assert getattr(row, f"field_{speed_field.id}") == 240
-    assert getattr(row, f"field_{price_field.id}") == 59999.99
+    assert getattr(row, f"field_{price_field.id}") == Decimal("59999.99")
     assert not getattr(row, "field_9999", None)
 
 
diff --git a/backend/tests/baserow/contrib/database/rows/test_rows_handler.py b/backend/tests/baserow/contrib/database/rows/test_rows_handler.py
index b00bdf97a1..7f8f7157b4 100644
--- a/backend/tests/baserow/contrib/database/rows/test_rows_handler.py
+++ b/backend/tests/baserow/contrib/database/rows/test_rows_handler.py
@@ -170,7 +170,7 @@ def test_create_row(send_mock, data_fixture):
     )
     assert getattr(row_1, f"field_{name_field.id}") == "Tesla"
     assert getattr(row_1, f"field_{speed_field.id}") == 240
-    assert getattr(row_1, f"field_{price_field.id}") == 59999.99
+    assert getattr(row_1, f"field_{price_field.id}") == Decimal("59999.99")
     assert not getattr(row_1, f"field_9999", None)
     assert row_1.order == 1
     row_1.refresh_from_db()
@@ -274,6 +274,118 @@ def test_create_row(send_mock, data_fixture):
     assert row_8.order == Decimal("3.00000000000000000000")
 
 
+@pytest.mark.django_db
+@pytest.mark.parametrize("force", [False, True])
+def test_create_row_prepared_values_skip_defaults_and_preparation(data_fixture, force):
+    user = data_fixture.create_user()
+    table = data_fixture.create_database_table(user=user)
+    status_field = data_fixture.create_single_select_field(table=table, name="Status")
+    status = data_fixture.create_select_option(field=status_field, value="Prepared")
+    tags_field = data_fixture.create_multiple_select_field(table=table, name="Tags")
+    default_tag = data_fixture.create_select_option(field=tags_field, value="Default")
+    FieldHandler().update_field(
+        user, tags_field, multiple_select_default=[default_tag.id]
+    )
+    handler = RowHandler()
+    values = {status_field.db_column: status}
+
+    with (
+        patch.object(
+            handler, "prepare_values_with_defaults", side_effect=AssertionError
+        ),
+        patch.object(handler, "prepare_values", side_effect=AssertionError),
+        patch.object(handler, "prepare_rows_in_bulk", side_effect=AssertionError),
+    ):
+        if force:
+            row = handler.force_create_row(
+                None, table, values, values_already_prepared=True
+            )
+        else:
+            row = handler.create_row(user, table, values, values_already_prepared=True)
+
+    assert getattr(row, status_field.db_column).id == status.id
+    assert list(getattr(row, tags_field.db_column).all()) == []
+    assert row.created_by == (None if force else user)
+    assert row.last_modified_by == (None if force else user)
+    assert values == {status_field.db_column: status}
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("force", [False, True])
+@pytest.mark.parametrize("user_field_names", [False, True])
+def test_create_row_preserves_field_key_handling(data_fixture, force, user_field_names):
+    user = data_fixture.create_user()
+    table = data_fixture.create_database_table(user=user)
+    name_field = data_fixture.create_text_field(
+        table=table, name="Name", text_default="Default name"
+    )
+    notes_field = data_fixture.create_text_field(
+        table=table, name="Notes", text_default="Default notes"
+    )
+    values = (
+        {"Name": "Explicit name"}
+        if user_field_names
+        else {
+            name_field.id: "Explicit name",
+            name_field.db_column: "Ignored duplicate",
+            "unknown": "Ignored",
+        }
+    )
+    handler = RowHandler()
+    create_row = handler.force_create_row if force else handler.create_row
+
+    row = create_row(user, table, values, user_field_names=user_field_names)
+
+    assert getattr(row, name_field.db_column) == "Explicit name"
+    assert getattr(row, notes_field.db_column) == "Default notes"
+
+
+@pytest.mark.django_db
+def test_create_row_preserves_signals_and_returns_refreshed_values(data_fixture):
+    user = data_fixture.create_user()
+    table = data_fixture.create_database_table(user=user)
+    name_field = data_fixture.create_text_field(table=table, name="Name")
+    formula_field = data_fixture.create_formula_field(
+        table=table, name="Formula", formula="field('Name') + '!'"
+    )
+    tags_field = data_fixture.create_multiple_select_field(table=table, name="Tags")
+    first = data_fixture.create_select_option(field=tags_field, value="First")
+    second = data_fixture.create_select_option(field=tags_field, value="Second")
+    handler = RowHandler()
+    before_row = handler.create_row(user, table, {name_field.db_column: "Existing"})
+
+    with (
+        patch(
+            "baserow.contrib.database.rows.signals.before_rows_create.send"
+        ) as before,
+        patch("baserow.contrib.database.rows.signals.rows_created.send") as created,
+    ):
+        row = handler.create_row(
+            user,
+            table,
+            {
+                name_field.db_column: "Created",
+                tags_field.db_column: [second.id, first.id],
+            },
+            before_row=before_row,
+            send_webhook_events=False,
+        )
+
+    assert row.order < before_row.order
+    assert getattr(row, formula_field.db_column) == "Created!"
+    assert [option.id for option in getattr(row, tags_field.db_column).all()] == [
+        second.id,
+        first.id,
+    ]
+    before.assert_called_once()
+    created.assert_called_once()
+    assert created.call_args.kwargs["before_return"] == before.return_value
+    assert created.call_args.kwargs["before"] is before_row
+    assert created.call_args.kwargs["send_webhook_events"] is False
+    assert created.call_args.kwargs["send_realtime_update"] is True
+    assert created.call_args.kwargs["rows"][0].id == row.id
+
+
 @pytest.mark.django_db
 def test_get_row(data_fixture):
     user = data_fixture.create_user()
@@ -703,6 +815,157 @@ def test_update_row_by_id(send_mock, data_fixture):
     assert send_mock.call_args[1]["before_return"] == before_send_mock.return_value
 
 
+@pytest.mark.django_db
+@pytest.mark.parametrize("by_id", [False, True])
+def test_update_row_preserves_field_key_handling(data_fixture, by_id):
+    user = data_fixture.create_user()
+    table = data_fixture.create_database_table(user=user)
+    name_field = data_fixture.create_text_field(table=table, name="Name")
+    notes_field = data_fixture.create_text_field(table=table, name="Notes")
+    handler = RowHandler()
+    row = handler.create_row(user, table, {name_field.id: "Original"})
+    other_row = handler.create_row(user, table, {name_field.id: "Other"})
+    values = {
+        name_field.id: "Integer key wins",
+        name_field.db_column: "Ignored duplicate",
+        notes_field.db_column: "Updated notes",
+        "unknown": "Ignored",
+        "field_999999999": "Ignored",
+        "id": other_row.id,
+    }
+    original_values = values.copy()
+
+    if by_id:
+        updated_row = handler.update_row_by_id(user, table, row.id, values)
+    else:
+        updated_row = handler.update_row(user, table, row, values)
+
+    assert updated_row.id == row.id
+    assert getattr(updated_row, name_field.db_column) == "Integer key wins"
+    assert getattr(updated_row, notes_field.db_column) == "Updated notes"
+    assert values == original_values
+    row.refresh_from_db()
+    other_row.refresh_from_db()
+    assert getattr(row, name_field.db_column) == "Integer key wins"
+    assert getattr(row, notes_field.db_column) == "Updated notes"
+    assert getattr(other_row, name_field.db_column) == "Other"
+
+
+@pytest.mark.django_db
+def test_update_row_refreshes_supplied_row_and_cached_relations(data_fixture):
+    user = data_fixture.create_user()
+    table = data_fixture.create_database_table(user=user)
+    name_field = data_fixture.create_text_field(table=table, name="Name")
+    formula_field = data_fixture.create_formula_field(
+        table=table, name="Formula", formula="field('Name') + '!'"
+    )
+    status_field = data_fixture.create_single_select_field(table=table, name="Status")
+    tags_field = data_fixture.create_multiple_select_field(table=table, name="Tags")
+    old_status = data_fixture.create_select_option(field=status_field, value="Old")
+    new_status = data_fixture.create_select_option(field=status_field, value="New")
+    old_tag = data_fixture.create_select_option(field=tags_field, value="Old")
+    new_tag = data_fixture.create_select_option(field=tags_field, value="New")
+    handler = RowHandler()
+    row = handler.create_row(
+        user,
+        table,
+        {
+            name_field.db_column: "Before",
+            status_field.db_column: old_status.id,
+            tags_field.db_column: [old_tag.id],
+        },
+    )
+    model = table.get_model()
+    row = model.objects.all().enhance_by_fields().get(id=row.id)
+    assert getattr(row, status_field.db_column).id == old_status.id
+    assert [option.id for option in getattr(row, tags_field.db_column).all()] == [
+        old_tag.id
+    ]
+    assert getattr(row, formula_field.db_column) == "Before!"
+
+    updated_row = handler.update_row(
+        user,
+        table,
+        row,
+        {
+            name_field.db_column: "After",
+            status_field.db_column: new_status.id,
+            tags_field.db_column: [new_tag.id],
+        },
+        model=model,
+    )
+
+    assert updated_row is row
+    assert getattr(row, name_field.db_column) == "After"
+    assert getattr(row, formula_field.db_column) == "After!"
+    assert getattr(row, status_field.db_column).id == new_status.id
+    assert [option.id for option in getattr(row, tags_field.db_column).all()] == [
+        new_tag.id
+    ]
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("by_id", [False, True])
+def test_update_row_accepts_prepared_values_without_preparing_again(
+    data_fixture, by_id
+):
+    user = data_fixture.create_user()
+    table = data_fixture.create_database_table(user=user)
+    field = data_fixture.create_single_select_field(table=table)
+    option = data_fixture.create_select_option(field=field, value="Prepared")
+    handler = RowHandler()
+    row = handler.create_row(user, table)
+    values = {field.db_column: option}
+
+    with (
+        patch.object(handler, "prepare_values", side_effect=AssertionError),
+        patch.object(handler, "prepare_rows_in_bulk", side_effect=AssertionError),
+    ):
+        if by_id:
+            updated_row = handler.update_row_by_id(
+                user, table, row.id, values, values_already_prepared=True
+            )
+        else:
+            updated_row = handler.update_row(
+                user, table, row, values, values_already_prepared=True
+            )
+
+    assert getattr(updated_row, field.db_column).id == option.id
+    row.refresh_from_db()
+    assert getattr(row, field.db_column).id == option.id
+    assert values == {field.db_column: option}
+
+
+@pytest.mark.django_db
+@patch("baserow.contrib.database.rows.signals.rows_updated.send")
+def test_update_row_tracks_prepared_link_models_as_ids(send_mock, data_fixture):
+    user = data_fixture.create_user()
+    table, linked_table, field = data_fixture.create_two_linked_tables(user=user)
+    handler = RowHandler()
+    first = handler.create_row(user, linked_table)
+    second = handler.create_row(user, linked_table)
+    row = handler.create_row(user, table, {field.db_column: [first.id]})
+
+    handler.update_row(
+        user,
+        table,
+        row,
+        {field.db_column: [first, second]},
+        values_already_prepared=True,
+    )
+
+    send_mock.assert_called_once()
+    tracker = send_mock.call_args.kwargs["m2m_change_tracker"]
+    deleted = tracker.get_deleted_m2m_rels_per_field_id_for_type("link_row")
+    created = tracker.get_created_m2m_rels_per_field_for_type("link_row")
+    assert deleted[field][row] == set()
+    assert created[field][row] == {second.id}
+    assert [linked.id for linked in getattr(row, field.db_column).all()] == [
+        first.id,
+        second.id,
+    ]
+
+
 @pytest.mark.django_db
 def test_update_row_last_modified_by(data_fixture):
     workspace = data_fixture.create_workspace()
@@ -2321,3 +2584,125 @@ def test_get_row_names_does_not_scale_queries_with_relational_primary(data_fixtu
         RowHandler().get_row_names(table, large_ids)
 
     assert len(small_captured.captured_queries) == len(large_captured.captured_queries)
+
+
+def _select_option_ids_on_a_hash_reversing_pair(data_fixture, field):
+    """
+    Creates two select options whose ids make ``{a, b}`` iterate as ``[b, a]``.
+
+    Django's m2m ``set`` bulk creates through rows while iterating a Python set, so
+    the stored order only diverges from the given list for ids that hash into
+    descending buckets. Roughly one in eight consecutive id pairs does.
+
+    :param data_fixture: The fixture used to create the select options.
+    :param field: The field the select options belong to.
+    :return: The two select options, in the order they must be written.
+    """
+
+    while True:
+        option = data_fixture.create_select_option(field=field, value="x", color="red")
+        if option.id % 8 == 6:
+            break
+
+    first = data_fixture.create_select_option(field=field, value="A", color="red")
+    second = data_fixture.create_select_option(field=field, value="B", color="blue")
+    assert list({first.id, second.id}) == [second.id, first.id]
+    return first, second
+
+
+@pytest.mark.django_db
+def test_update_row_by_id_keeps_multiple_select_option_order(data_fixture):
+    user = data_fixture.create_user()
+    table = data_fixture.create_database_table(user=user)
+    field = FieldHandler().create_field(
+        user=user, table=table, name="Tags", type_name="multiple_select"
+    )
+    first, second = _select_option_ids_on_a_hash_reversing_pair(data_fixture, field)
+
+    row = RowHandler().create_row(user=user, table=table)
+    RowHandler().update_row_by_id(
+        user, table, row.id, {field.db_column: [first.id, second.id]}
+    )
+
+    model = table.get_model()
+    stored = model.objects.prefetch_related(field.db_column).get(id=row.id)
+    assert [o.value for o in getattr(stored, field.db_column).all()] == ["A", "B"]
+
+
+@pytest.mark.django_db
+def test_update_row_by_id_removes_and_dedupes_multiple_select_options(data_fixture):
+    user = data_fixture.create_user()
+    table = data_fixture.create_database_table(user=user)
+    field = FieldHandler().create_field(
+        user=user, table=table, name="Tags", type_name="multiple_select"
+    )
+    first, second = _select_option_ids_on_a_hash_reversing_pair(data_fixture, field)
+    third = data_fixture.create_select_option(field=field, value="C", color="green")
+
+    row = RowHandler().create_row(
+        user=user, table=table, values={field.db_column: [first.id, second.id]}
+    )
+    model = table.get_model()
+
+    RowHandler().update_row_by_id(
+        user, table, row.id, {field.db_column: [second.id, third.id, second.id]}
+    )
+    stored = model.objects.prefetch_related(field.db_column).get(id=row.id)
+    assert [o.value for o in getattr(stored, field.db_column).all()] == ["B", "C"]
+
+    RowHandler().update_row_by_id(user, table, row.id, {field.db_column: []})
+    stored = model.objects.prefetch_related(field.db_column).get(id=row.id)
+    assert list(getattr(stored, field.db_column).all()) == []
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("update_in_bulk", [False, True])
+@pytest.mark.parametrize("self_referencing", [False, True])
+def test_update_row_keeps_links_to_trashed_rows(
+    data_fixture, update_in_bulk, self_referencing
+):
+    user = data_fixture.create_user()
+    table = data_fixture.create_database_table(user=user)
+    linked_table = (
+        table
+        if self_referencing
+        else data_fixture.create_database_table(database=table.database)
+    )
+    field = FieldHandler().create_field(
+        user=user,
+        table=table,
+        name="Links",
+        type_name="link_row",
+        link_row_table=linked_table,
+    )
+    handler = RowHandler()
+    first, second, third = [
+        handler.create_row(user=user, table=linked_table) for _ in range(3)
+    ]
+    row = handler.create_row(
+        user=user, table=table, values={field.db_column: [first.id, second.id]}
+    )
+    handler.delete_row(user, linked_table, second)
+
+    values = {field.db_column: [first.id, second.id, third.id]}
+    if update_in_bulk:
+        updated_row = handler.update_rows(
+            user, table, [{"id": row.id, **values}]
+        ).updated_rows[0]
+    else:
+        updated_row = handler.update_row_by_id(user, table, row.id, values)
+
+    assert [r.id for r in getattr(updated_row, field.db_column).all()] == [
+        first.id,
+        third.id,
+    ]
+
+    TrashHandler.restore_item(
+        user, "row", second.id, parent_trash_item_id=linked_table.id
+    )
+    stored = table.get_model().objects.prefetch_related(field.db_column).get(id=row.id)
+    assert [r.id for r in getattr(stored, field.db_column).all()] == [
+        first.id,
+        second.id,
+        third.id,
+    ]
diff --git a/backend/tests/baserow/contrib/database/workflow_actions/test_dispatch.py b/backend/tests/baserow/contrib/database/workflow_actions/test_dispatch.py
index e04c21f021..b1f9d69309 100644
--- a/backend/tests/baserow/contrib/database/workflow_actions/test_dispatch.py
+++ b/backend/tests/baserow/contrib/database/workflow_actions/test_dispatch.py
@@ -8,6 +8,9 @@
 from baserow.contrib.database.fields.handler import FieldHandler
 from baserow.contrib.database.rows.signals import rows_created
 from baserow.contrib.database.table.handler import TableHandler
+from baserow.contrib.database.workflow_actions.actions import (
+    DispatchButtonFieldActionType,
+)
 from baserow.contrib.database.workflow_actions.exceptions import (
     WorkflowActionDispatchError,
     WorkflowActionDispatchInProgress,
@@ -23,6 +26,7 @@
 from baserow.contrib.database.workflow_actions.service import (
     DatabaseWorkflowActionService,
 )
+from baserow.core.action.signals import action_done
 from baserow.core.exceptions import PermissionException
 
 
@@ -916,3 +920,108 @@ def test_a_later_action_reads_an_earlier_action_result(data_fixture):
     created = list(table.get_model().objects.exclude(id=row.id).order_by("id"))
     assert getattr(created[0], f"field_{name_field.id}") == "Ada"
     assert getattr(created[1], f"field_{name_field.id}") == f"Ada {created[0].id}"
+
+
+@pytest.fixture
+def audited_clicks():
+    """The `action_params` and workspace of every button click registration."""
+
+    received = []
+
+    def receiver(sender, action_type, action_params, workspace, **kwargs):
+        if action_type is DispatchButtonFieldActionType:
+            received.append((action_params, workspace))
+
+    action_done.connect(receiver)
+    yield received
+    action_done.disconnect(receiver)
+
+
+@pytest.mark.django_db
+def test_a_click_sends_action_done_for_the_audit_log(data_fixture, audited_clicks):
+    user = data_fixture.create_user()
+    table, name_field = _table_with_name(data_fixture, user)
+    button_field = data_fixture.create_button_field(table=table, label="Go")
+    row = table.get_model().objects.create()
+    _create_row_action(data_fixture, button_field, table, name_field, "first")
+
+    DatabaseWorkflowActionService().dispatch_workflow_actions(user, button_field, row)
+
+    assert len(audited_clicks) == 1
+    params, workspace = audited_clicks[0]
+    assert workspace == table.database.workspace
+    assert params["field_id"] == button_field.id
+    assert params["row_id"] == row.id
+    assert params["action_count"] == 1
+
+
+@pytest.mark.django_db
+def test_a_refused_click_sends_nothing_to_the_audit_log(data_fixture, audited_clicks):
+    owner = data_fixture.create_user()
+    outsider = data_fixture.create_user()
+    table, name_field = _table_with_name(data_fixture, owner)
+    button_field = data_fixture.create_button_field(table=table, label="Go")
+    row = table.get_model().objects.create()
+    _create_row_action(data_fixture, button_field, table, name_field, "first")
+
+    with pytest.raises(PermissionException):
+        DatabaseWorkflowActionService().dispatch_workflow_actions(
+            outsider, button_field, row
+        )
+
+    assert audited_clicks == []
+
+
+@pytest.mark.django_db
+def test_a_click_refused_for_a_run_in_progress_is_not_audited(
+    data_fixture, audited_clicks
+):
+    user = data_fixture.create_user()
+    table, name_field = _table_with_name(data_fixture, user)
+    button_field = data_fixture.create_button_field(table=table, label="Go")
+    row = table.get_model().objects.create()
+    _create_row_action(data_fixture, button_field, table, name_field, "first")
+    cache.add(f"button_dispatch_{button_field.id}_{row.id}", True, timeout=30)
+
+    with pytest.raises(WorkflowActionDispatchInProgress):
+        DatabaseWorkflowActionService().dispatch_workflow_actions(
+            user, button_field, row
+        )
+
+    assert audited_clicks == []
+
+
+@pytest.mark.django_db
+def test_a_click_with_only_client_actions_is_audited(data_fixture, audited_clicks):
+    user = data_fixture.create_user()
+    table, _ = _table_with_name(data_fixture, user)
+    button_field = data_fixture.create_button_field(table=table, label="Go")
+    row = table.get_model().objects.create()
+    data_fixture.create_database_workflow_action(
+        OpenUrlWorkflowAction, field=button_field
+    )
+
+    DatabaseWorkflowActionService().dispatch_workflow_actions(user, button_field, row)
+
+    assert len(audited_clicks) == 1
+
+
+@pytest.mark.django_db
+def test_a_click_that_fails_mid_sequence_is_still_audited(data_fixture, audited_clicks):
+    user = data_fixture.create_user()
+    table, name_field = _table_with_name(data_fixture, user)
+    button_field = data_fixture.create_button_field(table=table, label="Go")
+    row = table.get_model().objects.create()
+    _create_row_action(data_fixture, button_field, table, name_field, "first")
+    # A delete-row action with no table configured fails at dispatch.
+    data_fixture.create_database_workflow_action(
+        LocalBaserowDeleteRowWorkflowAction, field=button_field
+    )
+
+    with pytest.raises(WorkflowActionDispatchError):
+        DatabaseWorkflowActionService().dispatch_workflow_actions(
+            user, button_field, row
+        )
+
+    assert len(audited_clicks) == 1
+    assert audited_clicks[0][0]["action_count"] == 2
diff --git a/backend/tests/baserow/contrib/database/workflow_actions/test_start_workflow_action.py b/backend/tests/baserow/contrib/database/workflow_actions/test_start_workflow_action.py
index 29254f0c6a..4a3a48c9df 100644
--- a/backend/tests/baserow/contrib/database/workflow_actions/test_start_workflow_action.py
+++ b/backend/tests/baserow/contrib/database/workflow_actions/test_start_workflow_action.py
@@ -411,7 +411,7 @@ def test_a_click_starts_the_published_workflow(data_fixture):
     ) as async_start_workflow:
         DatabaseWorkflowActionService().dispatch_workflow_actions(user, field, row)
 
-    async_start_workflow.assert_called_once_with(published)
+    async_start_workflow.assert_called_once_with(published, triggered_by=user)
 
 
 @pytest.mark.django_db
@@ -459,6 +459,7 @@ def test_a_click_through_the_api_queues_the_published_workflow(
     history = AutomationWorkflowHistory.objects.get(original_workflow=workflow)
     assert history.workflow_id == published.id
     assert history.status == "started"
+    assert history.triggered_by_id == clicker.id
     celery_task.delay.assert_called_once_with(published.id, history.id)
 
 
diff --git a/backend/tests/baserow/contrib/integrations/core/test_core_periodic_service_type.py b/backend/tests/baserow/contrib/integrations/core/test_core_periodic_service_type.py
index fa8e985217..09b819fee2 100644
--- a/backend/tests/baserow/contrib/integrations/core/test_core_periodic_service_type.py
+++ b/backend/tests/baserow/contrib/integrations/core/test_core_periodic_service_type.py
@@ -322,6 +322,7 @@ def test_call_multiple_periodic_services_that_are_due(
                     "triggered_at": "2025-02-15T10:30:00+00:00",
                     "next_run_at": "2025-02-15T10:31:00+00:00",
                 },
+                triggered_by=None,
             ),
             call(
                 workflow_2,
@@ -329,6 +330,7 @@ def test_call_multiple_periodic_services_that_are_due(
                     "triggered_at": "2025-02-15T10:30:00+00:00",
                     "next_run_at": "2025-02-15T10:31:00+00:00",
                 },
+                triggered_by=None,
             ),
         ]
     )
diff --git a/backend/tests/baserow/contrib/integrations/core/test_core_start_workflow_service_type.py b/backend/tests/baserow/contrib/integrations/core/test_core_start_workflow_service_type.py
index dadf2342c4..466dcf6933 100644
--- a/backend/tests/baserow/contrib/integrations/core/test_core_start_workflow_service_type.py
+++ b/backend/tests/baserow/contrib/integrations/core/test_core_start_workflow_service_type.py
@@ -69,7 +69,7 @@ def test_start_workflow_service_dispatch_starts_configured_workflow(data_fixture
     ) as async_start_workflow:
         result = ServiceHandler().dispatch_service(service, fake_dispatch_context())
 
-    async_start_workflow.assert_called_once_with(published_workflow)
+    async_start_workflow.assert_called_once_with(published_workflow, triggered_by=None)
     assert result.data is None
 
 
@@ -125,7 +125,7 @@ def test_start_workflow_service_dispatch_starts_immediate_dispatch_workflow(
     ) as async_start_workflow:
         result = ServiceHandler().dispatch_service(service, fake_dispatch_context())
 
-    async_start_workflow.assert_called_once_with(published_workflow)
+    async_start_workflow.assert_called_once_with(published_workflow, triggered_by=None)
     assert result.data is None
 
 
@@ -220,3 +220,23 @@ def test_start_workflow_service_dispatch_without_workflow_raises(data_fixture):
 
     with pytest.raises(ServiceImproperlyConfiguredDispatchException):
         ServiceHandler().dispatch_service(service, fake_dispatch_context())
+
+
+@pytest.mark.django_db
+def test_start_workflow_service_dispatch_names_the_context_actor(data_fixture):
+    user = data_fixture.create_user()
+    workflow = data_fixture.create_automation_workflow(
+        user=user, trigger_type=CoreManualTriggerNodeType.type
+    )
+    published_workflow = AutomationWorkflowHandler().publish(workflow)
+    service = data_fixture.create_core_start_workflow_service(workflow=workflow)
+    dispatch_context = fake_dispatch_context()
+    dispatch_context.actor = user
+
+    with patch(
+        "baserow.contrib.automation.workflows.handler."
+        "AutomationWorkflowHandler.async_start_workflow"
+    ) as async_start_workflow:
+        ServiceHandler().dispatch_service(service, dispatch_context)
+
+    async_start_workflow.assert_called_once_with(published_workflow, triggered_by=user)
diff --git a/backend/tests/baserow/core/service/test_service_type.py b/backend/tests/baserow/core/service/test_service_type.py
index 7122b49e61..5aab77dbe5 100644
--- a/backend/tests/baserow/core/service/test_service_type.py
+++ b/backend/tests/baserow/core/service/test_service_type.py
@@ -369,7 +369,9 @@ def test_dispatch_context_actor_survives_clone(data_fixture):
 
 
 @pytest.mark.django_db
-def test_dispatch_context_actor_survives_clone_without_own_properties(data_fixture):
+def test_dispatch_context_without_actor_in_own_properties_clones_without_one(
+    data_fixture,
+):
     from baserow.test_utils.pytest_conftest import FakeDispatchContext
 
     class StrictDispatchContext(FakeDispatchContext):
@@ -380,10 +382,7 @@ class StrictDispatchContext(FakeDispatchContext):
         def __init__(self, context=None):
             super().__init__(context=context or {})
 
-    user = data_fixture.create_user()
-    other_user = data_fixture.create_user()
     dispatch_context = StrictDispatchContext()
-    dispatch_context.actor = user
+    dispatch_context.actor = data_fixture.create_user()
 
-    assert dispatch_context.clone().actor == user
-    assert dispatch_context.clone(actor=other_user).actor == other_user
+    assert dispatch_context.clone().actor is None
diff --git a/backend/tests/baserow/core/test_core_registry.py b/backend/tests/baserow/core/test_core_registry.py
index 5713d6d31d..358de53d76 100644
--- a/backend/tests/baserow/core/test_core_registry.py
+++ b/backend/tests/baserow/core/test_core_registry.py
@@ -9,6 +9,7 @@
     InstanceTypeAlreadyRegistered,
     InstanceTypeDoesNotExist,
 )
+from baserow.core.registries import subject_type_registry
 from baserow.core.registry import (
     CustomFieldsInstanceMixin,
     CustomFieldsRegistryMixin,
@@ -18,6 +19,7 @@
     ModelRegistryMixin,
     Registry,
 )
+from baserow.core.subjects import AnonymousUserSubjectType, UserSubjectType
 
 
 class FakeModel(object):
@@ -199,3 +201,23 @@ def test_get_serializer(data_fixture):
 
     serializer = registry.get_serializer(database, request=True)
     assert "order" in serializer.data
+
+
+@pytest.mark.django_db
+def test_get_subject_returns_the_stored_subject(data_fixture):
+    user = data_fixture.create_user()
+
+    assert subject_type_registry.get_subject(UserSubjectType.type, user.id) == user
+
+
+@pytest.mark.django_db
+def test_get_subject_returns_none_for_a_deleted_subject(data_fixture):
+    user = data_fixture.create_user()
+    user_id = user.id
+    user.delete()
+
+    assert subject_type_registry.get_subject(UserSubjectType.type, user_id) is None
+
+
+def test_get_subject_returns_none_for_a_type_not_stored_in_the_database():
+    assert subject_type_registry.get_subject(AnonymousUserSubjectType.type, 1) is None
diff --git a/changelog/entries/unreleased/bug/6091_keep_the_order_of_multiple_select_options_and_linked_rows_wh.json b/changelog/entries/unreleased/bug/6091_keep_the_order_of_multiple_select_options_and_linked_rows_wh.json
new file mode 100644
index 0000000000..0c70882be5
--- /dev/null
+++ b/changelog/entries/unreleased/bug/6091_keep_the_order_of_multiple_select_options_and_linked_rows_wh.json
@@ -0,0 +1,9 @@
+{
+    "type": "bug",
+    "message": "Fix row updates failing when they include an existing link to a trashed row.",
+    "issue_origin": "github",
+    "issue_number": 6091,
+    "domain": "database",
+    "bullet_points": [],
+    "created_at": "2026-09-15"
+}
diff --git a/changelog/entries/unreleased/bug/6094_fixes_messages_to_the_ai_assistant_silently_failing_to_send.json b/changelog/entries/unreleased/bug/6094_fixes_messages_to_the_ai_assistant_silently_failing_to_send.json
new file mode 100644
index 0000000000..7e2de89567
--- /dev/null
+++ b/changelog/entries/unreleased/bug/6094_fixes_messages_to_the_ai_assistant_silently_failing_to_send.json
@@ -0,0 +1,9 @@
+{
+    "type": "bug",
+    "message": "Fixes messages to the AI assistant silently failing to send",
+    "issue_origin": "github",
+    "issue_number": 6094,
+    "domain": "core",
+    "bullet_points": [],
+    "created_at": "2026-09-16"
+}
diff --git a/docs/decisions/006-button-field-workflow-actions.md b/docs/decisions/006-button-field-workflow-actions.md
index cb273640ca..65e0967154 100644
--- a/docs/decisions/006-button-field-workflow-actions.md
+++ b/docs/decisions/006-button-field-workflow-actions.md
@@ -314,9 +314,10 @@ down to the services:
 - No integration attached, which is every database button in v1: the service authorizes
   and executes as the actor, and fails if there is none.
 - Integration present, which is the builder, automation, and any future opt-in:
-  `authorized_user` authorizes, unchanged. Also recording the actor for auditing in this
-  branch is future work owned by the builder and automation teams, since today's
-  pipeline carries a single user; nothing in v1 depends on it.
+  `authorized_user` authorizes, unchanged. Who started the run is recorded beside that,
+  not instead of it: the run's history carries the person as `triggered_by`, and the
+  dispatch context's `actor` stays empty, so a node that happens to have no integration
+  does not start acting as whoever clicked.
 
 Whether "no integration attached" is modeled as a nullable foreign key on the service or
 as a small purpose-built object with the same interface is an implementation choice, not
@@ -405,6 +406,18 @@ for every other action type: the row actions, the HTTP request, the email and th
 message all act as the clicker or as this installation, and none of them borrows another
 user's reach.
 
+The click itself is not anonymous, though. The run's history records the clicker as a
+subject (`triggered_by_id`, `triggered_by_type` and `triggered_by_name`, so an agent can
+start a run later and the name survives the user being deleted), and the click is
+registered as a `dispatch_button_field` action, so the audit log holds who clicked which
+button on which row. None of that changes who the nodes act as: the context's `actor` slot, which a Local
+Baserow node without an integration would act as, stays empty for a run. A workflow that
+another workflow starts records nobody, since the person is not a member of wherever
+that second workflow lives. A test run or simulation started from the editor records
+whoever pressed the button, including when it waits for its trigger's event: the
+workflow keeps that subject beside its other temporary test state until the run starts
+or is cancelled.
+
 Not charging the button rate limit has one consequence worth stating plainly. When the
 automation module's own limits are what refuse a run, the clicker is not told:
 `async_start_workflow` catches the rate limit and the too-many-errors cases after dispatch
@@ -563,6 +576,13 @@ the natural place to narrow this further when it is wanted.
   and leaves the actions as they were saved. Builder workflow actions are the same, and
   making either undoable needs a way to restore a deleted action with its service, which
   neither has.
+- **Audit log.** A click registers one `dispatch_button_field` action after every
+  permission check has passed and before the first action runs, inside the lock when
+  there is one, so a click refused for permission or as already running leaves no entry
+  and a click that fails half way still does. A button with no actions returns before
+  registering anything, so its clicks leave no entry. A button with only client actions
+  takes no lock, so a double click on it is two entries. Clicks stay out of the undo
+  stack.
 - **Deleting a user.** Nothing breaks: actions run as whoever clicks, and v1 services
   have no integration, so no button depends on any particular account.
 - **Failure mid-sequence.** Execution stops, later actions are skipped, completed
diff --git a/e2e-tests/fixtures/automation/automationWorkflow.ts b/e2e-tests/fixtures/automation/automationWorkflow.ts
index c68d088281..9ae1f88138 100644
--- a/e2e-tests/fixtures/automation/automationWorkflow.ts
+++ b/e2e-tests/fixtures/automation/automationWorkflow.ts
@@ -1,4 +1,5 @@
 import { getClient } from "../../client"
+import { waitForJob } from "../job"
 import { Automation } from "./automation"
 
 export class AutomationWorkflow {
@@ -25,3 +26,18 @@ export async function createAutomationWorkflow(
     automation,
   )
 }
+
+/**
+ * Publishes a workflow and waits for the publish job to finish, so a button
+ * or a test run can start it the moment this returns.
+ */
+export async function publishAutomationWorkflow(
+  workflow: AutomationWorkflow,
+): Promise<void> {
+  const client = getClient(workflow.automation.workspace.user);
+  const job: any = await client.post(
+    `automation/workflows/${workflow.id}/publish/async/`,
+    {},
+  );
+  await waitForJob(client, job.data.id, `Publishing "${workflow.name}"`);
+}
diff --git a/e2e-tests/fixtures/database/field.ts b/e2e-tests/fixtures/database/field.ts
index 29f03dd2ea..ca2d4093c8 100644
--- a/e2e-tests/fixtures/database/field.ts
+++ b/e2e-tests/fixtures/database/field.ts
@@ -1,4 +1,5 @@
 import { getClient } from "../../client";
+import { waitForJob } from "../job";
 import { User } from "../user";
 import { Table } from "./table";
 
@@ -112,21 +113,10 @@ export async function duplicateField(
     { duplicate_data: options.copyData ?? false },
   );
 
-  const deadline = Date.now() + 30_000;
-  while (Date.now() < deadline) {
-    const poll: any = await client.get(`jobs/${job.data.id}/`);
-    if (poll.data.state === "failed") {
-      throw new Error(
-        `Duplicating "${field.name}" failed: ${
-          poll.data.human_readable_error || ""
-        }`,
-      );
-    }
-    if (poll.data.state === "finished") {
-      return poll.data.duplicated_field as Field;
-    }
-    await new Promise((resolve) => setTimeout(resolve, 500));
-  }
-
-  throw new Error(`Duplicating "${field.name}" did not finish in time`);
+  const finished = await waitForJob(
+    client,
+    job.data.id,
+    `Duplicating "${field.name}"`,
+  );
+  return finished.duplicated_field as Field;
 }
diff --git a/e2e-tests/fixtures/database/workflowAction.ts b/e2e-tests/fixtures/database/workflowAction.ts
index cf3e19abc8..d1369625ba 100644
--- a/e2e-tests/fixtures/database/workflowAction.ts
+++ b/e2e-tests/fixtures/database/workflowAction.ts
@@ -120,6 +120,18 @@ export async function createDeleteRowAction(
   });
 }
 
+/** An action that starts a published automation workflow by id. */
+export async function createStartWorkflowAction(
+  user: User,
+  buttonField: Field,
+  workflowId: number,
+): Promise<WorkflowAction> {
+  const action = await createWorkflowAction(user, buttonField, "start_workflow");
+  return updateWorkflowAction(user, action, {
+    service: { workflow_id: workflowId },
+  });
+}
+
 /** An action the browser runs itself, rather than the dispatch running it. */
 export async function createOpenUrlAction(
   user: User,
diff --git a/e2e-tests/fixtures/job.ts b/e2e-tests/fixtures/job.ts
new file mode 100644
index 0000000000..f62c555afe
--- /dev/null
+++ b/e2e-tests/fixtures/job.ts
@@ -0,0 +1,28 @@
+import { AxiosInstance } from "axios";
+
+/**
+ * Polls an async job until it finishes and returns its final payload, or
+ * throws with the job's own error when it fails or runs out of time.
+ */
+export async function waitForJob(
+  client: AxiosInstance,
+  jobId: number,
+  description: string,
+  { timeoutMs = 30_000, intervalMs = 500 } = {}
+): Promise<any> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    const poll: any = await client.get(`jobs/${jobId}/`);
+    if (poll.data.state === "failed") {
+      throw new Error(
+        `${description} failed: ${poll.data.human_readable_error || ""}`
+      );
+    }
+    if (poll.data.state === "finished") {
+      return poll.data;
+    }
+    await new Promise((resolve) => setTimeout(resolve, intervalMs));
+  }
+
+  throw new Error(`${description} did not finish in time`);
+}
diff --git a/e2e-tests/pages/baserowPage.ts b/e2e-tests/pages/baserowPage.ts
index 30aa064b71..28306f7446 100644
--- a/e2e-tests/pages/baserowPage.ts
+++ b/e2e-tests/pages/baserowPage.ts
@@ -22,6 +22,10 @@ export class BaserowPage {
   async authenticate(user: User) {
     await this.page.goto(`${this.baseUrl}?token=${user.refreshToken}`);
     await this.recoverFromNuxtError();
+    // The AI panel would otherwise open over the page header on every visit.
+    await this.page.evaluate(() => {
+      localStorage.setItem("baserow.rightSidebarOpen", "false");
+    });
   }
 
   async goto(params = {}) {
diff --git a/e2e-tests/pages/workspacePage.ts b/e2e-tests/pages/workspacePage.ts
index 13d7afc1fe..912f450c4c 100644
--- a/e2e-tests/pages/workspacePage.ts
+++ b/e2e-tests/pages/workspacePage.ts
@@ -17,8 +17,7 @@ export class WorkspacePage extends BaserowPage {
   }
 
   async authenticate() {
-    await this.page.goto(`${this.baseUrl}?token=${this.user.refreshToken}`);
-    await this.recoverFromNuxtError();
+    return super.authenticate(this.user);
   }
 
   getFullUrl() {
diff --git a/e2e-tests/tests/baserowTest.ts b/e2e-tests/tests/baserowTest.ts
index 06f8b21ea2..3c8fa76216 100644
--- a/e2e-tests/tests/baserowTest.ts
+++ b/e2e-tests/tests/baserowTest.ts
@@ -37,11 +37,6 @@ export const test = base.extend<BaserowFixtures>({
     const workspacePage = new WorkspacePage({ page, goto }, user, workspace);
     await workspacePage.authenticate();
 
-    await page.evaluate(() => {
-      // Prevent the AI panel to automatically open in all tests
-      localStorage.setItem("baserow.rightSidebarOpen", "false");
-    });
-
     // Use the fixture value in the test.
     await use(workspacePage);
 
diff --git a/e2e-tests/tests/database/button_field_start_workflow.spec.ts b/e2e-tests/tests/database/button_field_start_workflow.spec.ts
new file mode 100644
index 0000000000..791d1bcedb
--- /dev/null
+++ b/e2e-tests/tests/database/button_field_start_workflow.spec.ts
@@ -0,0 +1,179 @@
+/**
+ * Button field, start workflow action: a real click, the real worker, and a
+ * builder reading "Started by" in the History panel.
+ */
+
+import { Page } from "@playwright/test";
+import { test, expect } from "../baserowTest";
+import { GridPage } from "../../pages/database/gridPage";
+import { AutomationWorkflowPage } from "../../pages/automation/automationWorkflowPage";
+import { PageConfig } from "../../pages/baserowPage";
+import {
+  addAction,
+  openFieldEditor,
+  saveField,
+} from "../../pages/database/buttonFieldEditor";
+import { setupGrid, GridSetupResult } from "../../fixtures/database/gridSetup";
+import {
+  createStartWorkflowAction,
+  listWorkflowActions,
+} from "../../fixtures/database/workflowAction";
+import { createAutomation } from "../../fixtures/automation/automation";
+import {
+  AutomationWorkflow,
+  createAutomationWorkflow,
+  publishAutomationWorkflow,
+} from "../../fixtures/automation/automationWorkflow";
+import { createAutomationNode } from "../../fixtures/automation/automationNode";
+import { User, createUser } from "../../fixtures/user";
+import { addUserToWorkspace } from "../../fixtures/workspace";
+
+// Index of Start in the right-hand section; the first field `beforeAll` creates.
+const START_FIELD_INDEX = 0;
+
+let g: GridSetupResult;
+let workflow: AutomationWorkflow;
+/** A workflow only an event can start, so the picker must leave it out. */
+let eventWorkflow: AutomationWorkflow;
+/** A member of the workspace who can click the button but not open the automation. */
+let editor: User;
+
+/** The builder, who owns the automation, on the workflow page. */
+async function builderOnWorkflow(page: Page, goto: PageConfig["goto"]) {
+  const workflowPage = new AutomationWorkflowPage(
+    { page, goto },
+    workflow.automation,
+    workflow
+  );
+  await workflowPage.authenticate(g.user);
+  await workflowPage.goto();
+  return workflowPage;
+}
+
+async function openHistoryPanel(page: Page) {
+  await page.locator('[data-item-type="history"]').click();
+  await expect(page.locator(".history-side-panel__title")).toBeVisible();
+}
+
+/** The history entries in the panel, newest first, as the panel orders them. */
+function historyEntries(page: Page) {
+  return page.locator(".workflow-history__header");
+}
+
+test.describe("Button field, start workflow action", () => {
+  test.beforeAll(async () => {
+    g = await setupGrid({
+      dbName: "Start workflow DB",
+      tableName: "Jobs",
+      fields: [
+        { name: "Start", type: "button", settings: { label: "Start" } },
+        { name: "Configure", type: "button", settings: { label: "Configure" } },
+      ],
+      rows: [{ Name: "row one" }],
+    });
+
+    const automation = await createAutomation(
+      "Start workflow automation",
+      g.database.workspace
+    );
+    workflow = await createAutomationWorkflow("Started by button", automation);
+    await createAutomationNode(workflow, "manual");
+    await publishAutomationWorkflow(workflow);
+    eventWorkflow = await createAutomationWorkflow(
+      "Waits for rows",
+      automation
+    );
+    await createAutomationNode(eventWorkflow, "local_baserow_rows_created");
+    await createStartWorkflowAction(
+      g.user,
+      g.fieldByName["Start"],
+      workflow.id
+    );
+
+    editor = await createUser();
+    await addUserToWorkspace(g.user, g.database.workspace, editor, "MEMBER");
+  });
+
+  test("a click by an editor shows up as started by them in the workflow history", async ({
+    page,
+    browser,
+    goto,
+  }) => {
+    // The editor clicks, in their own browser.
+    const editorContext = await browser.newContext();
+    try {
+      const editorPage = await editorContext.newPage();
+      const grid = new GridPage(editorPage, editor);
+      await grid.goTo(g.database, g.table);
+      await grid.fieldCellAt(0, START_FIELD_INDEX).locator("button").click();
+      // The cell settles once the dispatch has returned.
+      await expect(
+        grid.fieldCellAt(0, START_FIELD_INDEX).locator("button")
+      ).toBeEnabled();
+    } finally {
+      await editorContext.close();
+    }
+
+    // The builder, who owns the automation, reads the run's history.
+    await builderOnWorkflow(page, goto);
+    await openHistoryPanel(page);
+
+    const entry = historyEntries(page).first();
+    await expect(
+      entry.locator("..").locator(".workflow-history__started-by")
+    ).toHaveText(`Started by ${editor.name}`);
+    // Not a test run: no prefix on the title. The worker in the e2e stack
+    // runs the workflow for real, so it also completes.
+    await expect(entry.locator(".workflow-history__header-title")).toHaveText(
+      "Ran successfully"
+    );
+  });
+
+  test("a test run from the editor is started by whoever pressed the button", async ({
+    page,
+    goto,
+  }) => {
+    await builderOnWorkflow(page, goto);
+
+    await page.locator('[data-highlight="automation-test-run"]').click();
+    await openHistoryPanel(page);
+
+    const entry = historyEntries(page).first();
+    await expect(
+      entry.locator(".workflow-history__header-title")
+    ).toContainText("[Test]");
+    await expect(
+      entry.locator("..").locator(".workflow-history__started-by")
+    ).toHaveText(`Started by ${g.user.name}`);
+  });
+
+  test("the editor offers only workflows a click can start, and saves the pick", async ({
+    page,
+  }) => {
+    const grid = new GridPage(page, g.user);
+    await grid.goTo(g.database, g.table);
+    await openFieldEditor(page, "Configure");
+    const added = await addAction(page, "Start workflow");
+
+    await added
+      .locator(".button-field-action-list__form .dropdown")
+      .first()
+      .click();
+    const items = page.locator(".dropdown__items:visible");
+    await expect(items).toContainText(workflow.name);
+    // A rows-created trigger cannot start on demand, so it is not on offer.
+    await expect(items).not.toContainText(eventWorkflow.name);
+    await items.getByText(workflow.name).click();
+
+    await saveField(page);
+    await expect(page.locator(".button-field-action-list")).toBeHidden();
+
+    const actions = await listWorkflowActions(
+      g.user,
+      g.fieldByName["Configure"]
+    );
+    const saved = actions.find((action) => action.type === "start_workflow");
+    expect(saved, "the start workflow action was not saved").toBeDefined();
+    expect(saved.service.workflow_id).toBe(workflow.id);
+  });
+});
diff --git a/e2e-tests/tests/database/realtimeReplay.spec.ts b/e2e-tests/tests/database/realtimeReplay.spec.ts
index 5718f4f293..022e769adf 100644
--- a/e2e-tests/tests/database/realtimeReplay.spec.ts
+++ b/e2e-tests/tests/database/realtimeReplay.spec.ts
@@ -73,10 +73,6 @@ async function setupRealtimeReplayScenario(
   const workspacePage = new WorkspacePage(pageConfig, user, workspace);
   await workspacePage.authenticate();
 
-  await page.evaluate(() => {
-    localStorage.setItem("baserow.rightSidebarOpen", "false");
-  });
-
   const tablePage = new TablePage(pageConfig);
   await tablePage.goToTable(table);
   await tablePage.waitForLoadingOverlayToDisappear();
diff --git a/enterprise/web-frontend/modules/baserow_enterprise/store/assistant.js b/enterprise/web-frontend/modules/baserow_enterprise/store/assistant.js
index d8aca9fd1b..a40f571da8 100644
--- a/enterprise/web-frontend/modules/baserow_enterprise/store/assistant.js
+++ b/enterprise/web-frontend/modules/baserow_enterprise/store/assistant.js
@@ -72,7 +72,7 @@ export const mutations = {
   },
 
   SET_CHATS(state, chats) {
-    state.chats = chats.map((chat) => ({
+    const fetched = chats.map((chat) => ({
       id: chat.uuid,
       title: chat.title,
       createdAt: chat.created_on,
@@ -84,6 +84,12 @@ export const mutations = {
       cancelling: false,
       currentMessageId: null,
     }))
+    // A chat created client-side is only persisted once its first message is
+    // sent, so the fetched list does not contain it yet.
+    const current = state.chats.find((c) => c.id === state.currentChatId)
+    const unsaved =
+      current && !fetched.some((c) => c.id === current.id) ? [current] : []
+    state.chats = [...unsaved, ...fetched]
   },
 
   SET_CHATS_LOADING(state, loading) {
@@ -91,7 +97,7 @@ export const mutations = {
   },
 
   REMOVE_CHAT(state, chatId) {
-    const index = state.chats.findIndex((chat) => chat.uid === chatId)
+    const index = state.chats.findIndex((chat) => chat.id === chatId)
     if (index > -1) {
       state.chats.splice(index, 1)
     }
@@ -260,10 +266,11 @@ export const actions = {
     { message, workspace }
   ) {
     const { $client, $i18n } = this
-    if (!state.currentChatId) {
+    let chat = state.chats.find((c) => c.id === state.currentChatId)
+    if (!chat) {
       await dispatch('createChat', workspace.id)
+      chat = state.chats.find((c) => c.id === state.currentChatId)
     }
-    const chat = state.chats.find((c) => c.id === state.currentChatId)
 
     const userMessage = {
       id: uuidv4(),
diff --git a/enterprise/web-frontend/test/unit/enterprise/store/assistant.spec.js b/enterprise/web-frontend/test/unit/enterprise/store/assistant.spec.js
new file mode 100644
index 0000000000..8c032ea832
--- /dev/null
+++ b/enterprise/web-frontend/test/unit/enterprise/store/assistant.spec.js
@@ -0,0 +1,128 @@
+import { beforeEach, describe, expect, test, vi } from 'vitest'
+
+import assistantService from '@baserow_enterprise/services/assistant'
+import { TestApp } from '@baserow/test/helpers/testApp'
+
+vi.mock('@baserow_enterprise/services/assistant', () => ({
+  default: vi.fn(),
+}))
+
+const workspace = { id: 1, name: 'Test workspace' }
+
+const serverChat = (uuid, title = 'Persisted') => ({
+  uuid,
+  title,
+  created_on: '2026-09-15T10:00:00Z',
+  updated_on: '2026-09-15T10:00:00Z',
+  status: 'completed',
+})
+
+describe('Assistant store', () => {
+  let testApp = null
+  let store = null
+  let service = null
+
+  beforeEach(async () => {
+    vi.clearAllMocks()
+    service = {
+      fetchChats: vi.fn().mockResolvedValue({ results: [] }),
+      fetchChatMessages: vi.fn().mockResolvedValue({ messages: [] }),
+      sendMessage: vi.fn().mockResolvedValue(undefined),
+      cancelMessage: vi.fn().mockResolvedValue(undefined),
+      submitFeedback: vi.fn().mockResolvedValue(undefined),
+    }
+    assistantService.mockReturnValue(service)
+
+    testApp = new TestApp()
+    store = testApp.store
+    // `uiContext` builds the assistant payload from the current undo/redo scope.
+    await store.dispatch('workspace/forceCreate', workspace)
+    await store.dispatch('undoRedo/updateCurrentScopeSet', {
+      workspace: workspace.id,
+    })
+  })
+
+  afterEach(async () => {
+    await testApp.afterEach()
+  })
+
+  test('sending a message recovers when the current chat is gone', async () => {
+    store.commit('assistant/SET_CURRENT_CHAT_ID', 'dropped')
+
+    await store.dispatch('assistant/sendMessage', {
+      message: 'hello',
+      workspace,
+    })
+
+    expect(service.sendMessage).toHaveBeenCalledOnce()
+    const currentChatId = store.getters['assistant/currentChatId']
+    expect(currentChatId).not.toBe('dropped')
+    expect(store.getters['assistant/currentChat']).toBeDefined()
+    expect(service.sendMessage.mock.calls[0][0]).toBe(currentChatId)
+  })
+
+  test('removing a chat removes it by id', async () => {
+    store.commit('assistant/SET_CHATS', [
+      serverChat('first'),
+      serverChat('second'),
+    ])
+
+    store.commit('assistant/REMOVE_CHAT', 'first')
+
+    expect(store.getters['assistant/chats'].map((c) => c.id)).toStrictEqual([
+      'second',
+    ])
+  })
+
+  test('sending a message leaves the assistant idle once it finishes', async () => {
+    await store.dispatch('assistant/createChat', workspace.id)
+    service.sendMessage.mockImplementation(
+      async (chatUuid, message, uiContext, onUpdate) => {
+        await onUpdate({ type: 'ai/message', content: 'hi' })
+      }
+    )
+
+    await store.dispatch('assistant/sendMessage', {
+      message: 'hello',
+      workspace,
+    })
+
+    expect(store.getters['assistant/currentChat'].running).toBe(false)
+  })
+
+  test('fetching chats keeps a current chat that is not persisted yet', async () => {
+    await store.dispatch('assistant/createChat', workspace.id)
+    const unsavedChatId = store.getters['assistant/currentChatId']
+    service.fetchChats.mockResolvedValue({ results: [serverChat('persisted')] })
+
+    await store.dispatch('assistant/fetchChats', workspace.id)
+
+    expect(store.getters['assistant/chats'].map((c) => c.id)).toStrictEqual([
+      unsavedChatId,
+      'persisted',
+    ])
+    expect(store.getters['assistant/currentChat']).toBeDefined()
+  })
+
+  test('fetching chats replaces a current chat the server also returns', async () => {
+    store.commit('assistant/SET_CHATS', [serverChat('persisted', 'Old title')])
+    store.commit('assistant/SET_CURRENT_CHAT_ID', 'persisted')
+    service.fetchChats.mockResolvedValue({
+      results: [serverChat('persisted', 'New title')],
+    })
+
+    await store.dispatch('assistant/fetchChats', workspace.id)
+
+    expect(store.getters['assistant/chats']).toHaveLength(1)
+    expect(store.getters['assistant/currentChat'].title).toBe('New title')
+  })
+
+  test('resetting clears the chats instead of keeping the current one', async () => {
+    await store.dispatch('assistant/createChat', workspace.id)
+
+    await store.dispatch('assistant/reset')
+
+    expect(store.getters['assistant/chats']).toStrictEqual([])
+    expect(store.getters['assistant/currentChatId']).toBe(null)
+  })
+})
diff --git a/web-frontend/modules/automation/components/workflow/sidePanels/WorkflowHistory.vue b/web-frontend/modules/automation/components/workflow/sidePanels/WorkflowHistory.vue
index 6c04f224d4..04c034c327 100644
--- a/web-frontend/modules/automation/components/workflow/sidePanels/WorkflowHistory.vue
+++ b/web-frontend/modules/automation/components/workflow/sidePanels/WorkflowHistory.vue
@@ -21,6 +21,9 @@
           type="secondary"
         />
       </div>
+      <div v-if="triggeredByName" class="workflow-history__started-by">
+        {{ $t('historySidePanel.startedBy', { name: triggeredByName }) }}
+      </div>
     </template>
 
     <template #default>
@@ -79,6 +82,7 @@ import historySuccessIcon from '@baserow/modules/core/assets/images/history-succ
 import historyFailedIcon from '@baserow/modules/core/assets/images/history-failed.svg?url'
 import historyDisabledIcon from '@baserow/modules/core/assets/images/history-disabled.svg?url'
 import NodeHistory from '@baserow/modules/automation/components/workflow/sidePanels/NodeHistory.vue'
+import { getCollaboratorName } from '@baserow/modules/core/utils/collaborator'
 
 const app = useNuxtApp()
 const store = useStore()
@@ -128,6 +132,15 @@ const nodeHistoriesEntry = computed(() =>
   store.getters['automationHistory/getNodeHistories'](props.item.id)
 )
 
+// A user's name comes from the workspace store so a rename shows; any other
+// subject, or a user who left, falls back to the name stored on the run.
+const triggeredByName = computed(() => {
+  const triggeredBy = props.item.triggered_by
+  if (!triggeredBy) return ''
+  if (triggeredBy.type !== 'auth.User') return triggeredBy.name
+  return getCollaboratorName(triggeredBy, store)
+})
+
 const statusTitle = computed(() => {
   switch (props.item.status) {
     case 'success':
diff --git a/web-frontend/modules/automation/locales/en.json b/web-frontend/modules/automation/locales/en.json
index 403414593a..2e8a4895e3 100644
--- a/web-frontend/modules/automation/locales/en.json
+++ b/web-frontend/modules/automation/locales/en.json
@@ -110,6 +110,7 @@
     "statusErrorBadge": "Failed",
     "statusSkippedBadge": "Skipped",
     "statusDisabled": "Disabled by Baserow",
+    "startedBy": "Started by {name}",
     "successfulRuns": "Successful runs",
     "completedInSeconds": "Completed in {s} seconds.",
     "completedInLessThanSecond": "Completed in less than 1 second.",
diff --git a/web-frontend/modules/core/assets/scss/components/automation/workflow/workflow_history.scss b/web-frontend/modules/core/assets/scss/components/automation/workflow/workflow_history.scss
index dc188a673c..55a0357ad8 100644
--- a/web-frontend/modules/core/assets/scss/components/automation/workflow/workflow_history.scss
+++ b/web-frontend/modules/core/assets/scss/components/automation/workflow/workflow_history.scss
@@ -26,6 +26,16 @@
   font-size: 11px;
 }
 
+.workflow-history__started-by {
+  margin-left: 26px;
+  margin-top: 4px;
+  font-size: 11px;
+  color: $palette-neutral-700;
+  overflow: hidden;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+}
+
 .workflow-history__run-time {
   margin-left: 24px;
   margin-top: 12px;
diff --git a/web-frontend/modules/core/utils/collaborator.js b/web-frontend/modules/core/utils/collaborator.js
new file mode 100644
index 0000000000..19e7532be1
--- /dev/null
+++ b/web-frontend/modules/core/utils/collaborator.js
@@ -0,0 +1,29 @@
+/**
+ * Returns the display name of a collaborator value, preferring the name in the
+ * workspace store so a rename shows without a reload.
+ */
+export function getCollaboratorName(collaboratorValue, store) {
+  // Workaround for field conversion not to produce console errors
+  if (
+    !collaboratorValue ||
+    typeof collaboratorValue !== 'object' ||
+    (!collaboratorValue.id && !collaboratorValue.name)
+  ) {
+    return ''
+  }
+
+  // If workspaces are unavailable, public views are served
+  const workspaces = store.getters['workspace/getAll']
+  if (workspaces.length === 0) {
+    return collaboratorValue.name
+  }
+
+  // Otherwise, get name from the store to reflect real-time updates
+  const user = store.getters['workspace/getUserById'](collaboratorValue.id)
+  if (user) {
+    return user.name
+  }
+
+  // Fallback if for some reason the user is missing from the store
+  return collaboratorValue.name
+}
diff --git a/web-frontend/modules/database/assets/scss/components/button_field_action_list.scss b/web-frontend/modules/database/assets/scss/components/button_field_action_list.scss
index a3b7cfeafb..aaf0be3ff4 100644
--- a/web-frontend/modules/database/assets/scss/components/button_field_action_list.scss
+++ b/web-frontend/modules/database/assets/scss/components/button_field_action_list.scss
@@ -71,6 +71,10 @@
 // padding of its own, does not run into the card's edges.
 .button-field-action-list__form {
   padding: 12px;
+
+  // Same cap as the error and payload below: a service form's helper text is
+  // a full sentence, and the popup would otherwise grow to fit it on one line.
+  max-width: 390px;
 }
 
 // Without this the control's inactive fill disappears into the card behind it.
diff --git a/web-frontend/modules/database/mixins/collaboratorName.js b/web-frontend/modules/database/mixins/collaboratorName.js
index cb9efdcb60..bb3c37e6c0 100644
--- a/web-frontend/modules/database/mixins/collaboratorName.js
+++ b/web-frontend/modules/database/mixins/collaboratorName.js
@@ -1,33 +1,9 @@
+import { getCollaboratorName } from '@baserow/modules/core/utils/collaborator'
+
 export default {
   methods: {
     getCollaboratorName(collaboratorValue, store) {
-      if (store === undefined) {
-        store = this.$store
-      }
-
-      // Workaround for field conversion not to produce console errors
-      if (
-        !collaboratorValue ||
-        typeof collaboratorValue !== 'object' ||
-        (!collaboratorValue.id && !collaboratorValue.name)
-      ) {
-        return ''
-      }
-
-      // If workspaces are unavailable, public views are served
-      const workspaces = store.getters['workspace/getAll']
-      if (workspaces.length === 0) {
-        return collaboratorValue.name
-      }
-
-      // Otherwise, get name from the store to reflect real-time updates
-      const user = store.getters['workspace/getUserById'](collaboratorValue.id)
-      if (user) {
-        return user.name
-      }
-
-      // Fallback if for some reason the user is missing from the store
-      return collaboratorValue.name
+      return getCollaboratorName(collaboratorValue, store ?? this.$store)
     },
     getCollaboratorNameInitials(collaboratorValue, store) {
       return this.getCollaboratorName(collaboratorValue, store)
diff --git a/web-frontend/modules/integrations/locales/en.json b/web-frontend/modules/integrations/locales/en.json
index 8951b94529..04ad6f5a7c 100644
--- a/web-frontend/modules/integrations/locales/en.json
+++ b/web-frontend/modules/integrations/locales/en.json
@@ -251,7 +251,7 @@
   "coreStartWorkflowServiceForm": {
     "automation": "Automation",
     "workflow": "Workflow",
-    "workflowHelperText": "Only published workflows with a manual trigger can be started. If this workflow has unpublished changes, the latest published version will run."
+    "workflowHelperText": "Only published workflows whose trigger can start on demand, such as a manual or periodic trigger, can be started. If this workflow has unpublished changes, the latest published version will run."
   },
   "coreGotoServiceForm": {
     "conditionLabel": "Condition",
diff --git a/web-frontend/test/unit/automation/components/workflow/WorkflowHistory.spec.js b/web-frontend/test/unit/automation/components/workflow/WorkflowHistory.spec.js
new file mode 100644
index 0000000000..3dcd72f48e
--- /dev/null
+++ b/web-frontend/test/unit/automation/components/workflow/WorkflowHistory.spec.js
@@ -0,0 +1,81 @@
+import { mountSuspended } from '@nuxt/test-utils/runtime'
+import WorkflowHistory from '@baserow/modules/automation/components/workflow/sidePanels/WorkflowHistory.vue'
+
+// WorkflowHistory reads the node histories from the automationHistory store
+// and dispatches fetches on toggle; neither matters for the header, so the
+// store is a stub. See WorkflowEdge.spec.js for why only useStore is mocked.
+const storeHolder = vi.hoisted(() => ({ store: null }))
+vi.mock('vuex', async (importOriginal) => ({
+  ...(await importOriginal()),
+  useStore: () => storeHolder.store,
+}))
+
+const mountHistory = (item, usersById = {}) => {
+  storeHolder.store = {
+    dispatch: vi.fn(),
+    getters: {
+      'automationHistory/getNodeHistories': () => null,
+      'workspace/getAll': [{ id: 1 }],
+      'workspace/getUserById': (id) => usersById[id] ?? null,
+    },
+  }
+  return mountSuspended(WorkflowHistory, {
+    props: { item },
+    global: {
+      stubs: { NodeHistory: true },
+      mocks: { $t: (key, params) => `${key}:${params?.name}` },
+    },
+  })
+}
+
+const baseItem = {
+  id: 1,
+  status: 'success',
+  started_on: '2026-09-09T10:00:00Z',
+  completed_on: '2026-09-09T10:00:02Z',
+  is_test_run: false,
+  message: '',
+  triggered_by: null,
+}
+
+describe('WorkflowHistory', () => {
+  afterEach(() => {
+    storeHolder.store = null
+  })
+
+  // `$t` is mocked to echo the name, so the assertions see which one is used.
+  test('names who started the run from the workspace store', async () => {
+    const wrapper = await mountHistory(
+      { ...baseItem, triggered_by: { id: 7, type: 'auth.User', name: 'Ada' } },
+      { 7: { id: 7, name: 'Ada Lovelace' } }
+    )
+    expect(wrapper.find('.workflow-history__started-by').text()).toBe(
+      'historySidePanel.startedBy:Ada Lovelace'
+    )
+  })
+
+  test('still names a user who left the workspace', async () => {
+    const wrapper = await mountHistory({
+      ...baseItem,
+      triggered_by: { id: 7, type: 'auth.User', name: 'Ada' },
+    })
+    expect(wrapper.find('.workflow-history__started-by').text()).toBe(
+      'historySidePanel.startedBy:Ada'
+    )
+  })
+
+  test('names a subject that is not a user from the run', async () => {
+    const wrapper = await mountHistory(
+      { ...baseItem, triggered_by: { id: 7, type: 'agent', name: 'Helper' } },
+      { 7: { id: 7, name: 'Ada Lovelace' } }
+    )
+    expect(wrapper.find('.workflow-history__started-by').text()).toBe(
+      'historySidePanel.startedBy:Helper'
+    )
+  })
+
+  test('shows nothing when nobody is recorded', async () => {
+    const wrapper = await mountHistory(baseItem)
+    expect(wrapper.find('.workflow-history__started-by').exists()).toBe(false)
+  })
+})