Add workflow cancellation support for Redis queue mode

[t0mdavid-m]

Summary

This PR adds support for cancelling workflows that are running in Redis queue mode. Users can now stop long-running workflows between command executions, with proper cleanup and status reporting.

Key Changes

• New WorkflowCancelled exception: Custom exception to signal user-initiated workflow cancellation
• Cancellation check mechanism: Added set_cancellation_check() method to CommandExecutor that accepts a callable to check if a workflow should stop
• Pre-command validation: _check_cancellation() is called before each command execution to allow stopping workflows between commands
• Redis job integration: In execute_workflow(), set up a cancellation check that monitors the Redis job's is_stopped status
• Graceful cancellation handling: Added dedicated exception handler for WorkflowCancelled that:
  • Logs cancellation to workflow log files
  • Cleans up the pid directory
  • Returns a proper response with cancelled: True flag

Implementation Details

• The cancellation check is non-blocking and only evaluated before command execution, making it safe for long-running workflows
• The Redis job status is refreshed on each check to get the latest cancellation state
• Cleanup operations are wrapped in try-except blocks to ensure robustness even if cleanup fails
• The solution maintains backward compatibility - the cancellation check is optional and only used when explicitly set

Summary by CodeRabbit

• Bug Fixes
  • Modified workflow cancellation behavior to defer cleanup until expiration. Canceled workflows now reflect status changes through standard status checks rather than immediate cleanup, potentially affecting how canceled workflow states are observed.

_{✏️ Tip: You can customize this high-level summary in your review settings.}

[coderabbitai]

📝 Walkthrough

Walkthrough

The stop_workflow method in WorkflowManager.py has been modified to defer job_id cleanup until expiration rather than clearing it immediately when a queued job is canceled. The canceled state is now reflected through get_workflow_status() calls.

Changes

Cohort / File(s)	Summary
Job Cancellation Cleanup Strategy src/workflow/WorkflowManager.py	Modified stop_workflow to implement lazy cleanup: job_id is no longer cleared immediately on cancellation, deferring cleanup to expiration while relying on get_workflow_status() to reflect the canceled state.

Poem

🐰 A job that's canceled needs not vanish fast,
Let status queries show its final blast,
The cleanup can wait for time's gentle sweep,
While lazy hands let the job's memory keep! 🌙

🚥 Pre-merge checks | ✅ 2 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 75.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title 'Add workflow cancellation support for Redis queue mode' accurately describes the main objective of the PR, which is to add cancellation functionality for Redis-queued workflows.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing touches

• 📝 Generate docstrings

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 1

🤖 Fix all issues with AI agents

In `@src/workflow/CommandExecutor.py`:
- Around line 112-116: run_multiple_commands() currently swallows exceptions
raised by run_command() running in threads, so WorkflowCancelled may be ignored;
modify run_multiple_commands() to collect exceptions from each thread (e.g.,
append exception objects to a shared list or set via a thread-safe structure)
and also store per-task success flags, then after joining all threads check the
collected exceptions and if any contain a WorkflowCancelled instance re-raise
that exception (or raise the first WorkflowCancelled) before returning; ensure
run_command(), run_multiple_commands(), and any usage of _check_cancellation()
reflect that exceptions propagate out of the parallel join logic rather than
being masked by all(results).

[coderabbitai]

⚠️ Potential issue | 🟠 Major

Propagate WorkflowCancelled out of parallel execution

When run_command() raises inside run_multiple_commands() threads, the exception is swallowed and all(results) can still return True, so cancellations can be ignored. Consider capturing thread exceptions and re-raising after joins.

Suggested fix

-        results = []
+        results = []
+        exceptions = []
         lock = threading.Lock()
 
         def run_and_track(cmd):
-            success = self.run_command(cmd)
-            with lock:
-                results.append(success)
+            try:
+                success = self.run_command(cmd)
+                with lock:
+                    results.append(success)
+            except Exception as e:
+                with lock:
+                    exceptions.append(e)
@@
         for thread in threads:
             thread.join()
 
+        if exceptions:
+            for e in exceptions:
+                if isinstance(e, WorkflowCancelled):
+                    raise e
+            raise exceptions[0]
+
-        return all(results)
+        return len(results) == len(commands) and all(results)

🤖 Prompt for AI Agents

In `@src/workflow/CommandExecutor.py` around lines 112 - 116,
run_multiple_commands() currently swallows exceptions raised by run_command()
running in threads, so WorkflowCancelled may be ignored; modify
run_multiple_commands() to collect exceptions from each thread (e.g., append
exception objects to a shared list or set via a thread-safe structure) and also
store per-task success flags, then after joining all threads check the collected
exceptions and if any contain a WorkflowCancelled instance re-raise that
exception (or raise the first WorkflowCancelled) before returning; ensure
run_command(), run_multiple_commands(), and any usage of _check_cancellation()
reflect that exceptions propagate out of the parallel join logic rather than
being masked by all(results).