feat: add active workflow cancellation (#990)

Author: paullizer

application/single_app/background_tasks.py

@@ -47,7 +47,7 @@
     update_group_workflow_runtime_fields,
 )
 from functions_settings import get_settings, is_group_workflows_enabled_for_group, update_settings
-from functions_workflow_runner import run_group_workflow, run_personal_workflow
+from functions_workflow_runner import create_workflow_run_id, run_group_workflow, run_personal_workflow
 
 
 def _get_lock_holder_id():
@@ -514,18 +514,26 @@ def check_due_workflows_once():
                         pass
 
                 started_at = datetime.now(timezone.utc).isoformat()
+                active_run_id = create_workflow_run_id()
                 update_personal_workflow_runtime_fields(
                     user_id,
                     workflow_id,
                     {
                         'status': 'running',
+                        'active_run_id': active_run_id,
+                        'cancellation_requested_at': None,
+                        'cancellation_requested_by': '',
                         'last_run_started_at': started_at,
                         'last_run_trigger_source': trigger_source,
                         'last_run_error': '',
                     },
                 )
 
-                result = run_personal_workflow(refreshed_workflow, trigger_source=trigger_source)
+                result = run_personal_workflow(
+                    refreshed_workflow,
+                    trigger_source=trigger_source,
+                    run_id=active_run_id,
+                )
                 update_fields = dict(result.get('workflow_updates') or {})
                 update_fields['status'] = 'idle'
                 update_fields['next_run_at'] = compute_next_run_at(refreshed_workflow, from_time=datetime.now(timezone.utc))
@@ -590,18 +598,26 @@ def check_due_workflows_once():
                         pass
 
                 started_at = datetime.now(timezone.utc).isoformat()
+                active_run_id = create_workflow_run_id()
                 update_group_workflow_runtime_fields(
                     group_id,
                     workflow_id,
                     {
                         'status': 'running',
+                        'active_run_id': active_run_id,
+                        'cancellation_requested_at': None,
+                        'cancellation_requested_by': '',
                         'last_run_started_at': started_at,
                         'last_run_trigger_source': trigger_source,
                         'last_run_error': '',
                     },
                 )
 
-                result = run_group_workflow(refreshed_workflow, trigger_source=trigger_source)
+                result = run_group_workflow(
+                    refreshed_workflow,
+                    trigger_source=trigger_source,
+                    run_id=active_run_id,
+                )
                 update_fields = dict(result.get('workflow_updates') or {})
                 update_fields['status'] = 'idle'
                 update_fields['next_run_at'] = compute_next_run_at(refreshed_workflow, from_time=datetime.now(timezone.utc))

application/single_app/config.py

@@ -97,7 +97,7 @@
 EXECUTOR_TYPE = 'thread'
 EXECUTOR_MAX_WORKERS = 30
 SESSION_TYPE = 'filesystem'
-VERSION = "0.250.061"
+VERSION = "0.250.062"
 IS_DEVELOPMENT = is_development_env_enabled()
 
 SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax')

application/single_app/functions_group_workflows.py

@@ -506,6 +506,9 @@ def save_group_workflow(group_id, workflow_data, actor_user_id, user_info=None):
         'last_run_response_preview': (existing_workflow or {}).get('last_run_response_preview', ''),
         'last_run_trigger_source': (existing_workflow or {}).get('last_run_trigger_source', ''),
         'run_count': int((existing_workflow or {}).get('run_count') or 0),
+        'active_run_id': (existing_workflow or {}).get('active_run_id', ''),
+        'cancellation_requested_at': (existing_workflow or {}).get('cancellation_requested_at'),
+        'cancellation_requested_by': (existing_workflow or {}).get('cancellation_requested_by', ''),
     }
 
     if trigger_type in {'interval', 'file_sync'} and is_enabled:

application/single_app/functions_personal_workflows.py

@@ -634,6 +634,9 @@ def save_personal_workflow(user_id, workflow_data, actor_user_id=None):
         'last_run_response_preview': (existing_workflow or {}).get('last_run_response_preview', ''),
         'last_run_trigger_source': (existing_workflow or {}).get('last_run_trigger_source', ''),
         'run_count': int((existing_workflow or {}).get('run_count') or 0),
+        'active_run_id': (existing_workflow or {}).get('active_run_id', ''),
+        'cancellation_requested_at': (existing_workflow or {}).get('cancellation_requested_at'),
+        'cancellation_requested_by': (existing_workflow or {}).get('cancellation_requested_by', ''),
     }
 
     if trigger_type in {'interval', 'file_sync'} and is_enabled:

application/single_app/functions_workflow_activity.py

@@ -34,7 +34,9 @@ def _normalize_status(value):
     normalized_value = _normalize_text(value).lower()
     if normalized_value in {'running', 'pending', 'in_progress', 'in-progress'}:
         return 'running'
-    if normalized_value in {'failed', 'error', 'cancelled', 'canceled'}:
+    if normalized_value in {'cancelled', 'canceled'}:
+        return 'cancelled'
+    if normalized_value in {'failed', 'error'}:
         return 'failed'
     if normalized_value in {'completed', 'complete', 'succeeded', 'success', 'done'}:
         return 'completed'
@@ -346,5 +348,5 @@ def build_workflow_activity_snapshot(run_record=None, workflow=None, conversatio
         'run': _serialize_run(run_record),
         'activities': activities,
         'lane_count': max(1, len(lane_order) or 1),
-        'live': _normalize_status((run_record or {}).get('status')) == 'running',
+        'live': _normalize_text((run_record or {}).get('status')).lower() in {'running', 'cancelling'},
     }