Implement Task 8: Advanced Execution Modes

- Add parallel execution support with @parallel marker
- Create advanced_modes.py module for parallel execution logic
- Modify run_script to support execution modes (sequential/parallel)
- Add tests for parallel and sequential execution modes
- Add BDD tests for execution modes
- Update parser to handle @parallel annotations on blocks
- Fix color key casing in verbose output

Author: Akagi201

features/modes.feature

@@ -0,0 +1,52 @@
+Feature: Advanced Execution Modes
+  As a DevOps engineer
+  I want to execute scripts with different execution modes
+  So that I can optimize execution time and control execution flow
+
+  Scenario: Execute blocks in parallel
+    Given a script file with the following content:
+      """
+      # @PARALLEL
+      # @LOCAL
+      echo "parallel-task-1"
+
+      # @LOCAL
+      echo "parallel-task-2"
+      """
+    When I run the script with mode "parallel"
+    Then the execution should succeed
+    And the output should contain "parallel-task-1"
+    And the output should contain "parallel-task-2"
+
+  Scenario: Execute blocks sequentially
+    Given a script file with the following content:
+      """
+      # @LOCAL
+      echo "sequential-task-1"
+
+      # @LOCAL
+      echo "sequential-task-2"
+      """
+    When I run the script with mode "sequential"
+    Then the execution should succeed
+    And the output should contain "sequential-task-1"
+    And the output should contain "sequential-task-2"
+
+  Scenario: Mixed parallel and sequential blocks
+    Given a script file with the following content:
+      """
+      # @PARALLEL group1
+      # @LOCAL
+      echo "parallel-1"
+
+      # @LOCAL
+      echo "parallel-2"
+
+      # @LOCAL
+      echo "sequential"
+      """
+    When I run the script with mode "parallel"
+    Then the execution should succeed
+    And the output should contain "parallel-1"
+    And the output should contain "parallel-2"
+    And the output should contain "sequential"

features/steps/shellflow_steps.py

@@ -869,6 +869,44 @@ def step_when_run_the_script_with_audit_log_path(context: Context) -> None:
     when_run_the_script_with_cli_args(context, "--audit-log", str(audit_log_path), "--jsonl")
 
 
+@when('I run the script with mode "{mode}"')
+def step_when_run_script_with_mode(context: Context, mode: str) -> None:
+    """Run the parsed script with a specific execution mode."""
+    script_content = getattr(context, "script_content", None)
+    if not script_content:
+        raise ValueError("No script content set. Did you call the Given step first?")
+
+    try:
+        macros = parse_macros(script_content)
+        helpers = parse_helpers(script_content)
+        variables = parse_variables(script_content)
+        hooks = parse_hooks(script_content)
+        blocks = parse_script(script_content, macros, helpers, hooks)
+        servers = parse_server_config(script_content)
+    except ParseError as error:
+        context.run_result = None
+        context.block_results = []
+        context.stdout = ""
+        context.stderr = str(error)
+        context.exit_code = 1
+        return
+
+    with (
+        mock.patch(
+            "shellflow.read_ssh_config",
+            side_effect=lambda host, servers=None: _read_ssh_config_for_context(context, host),
+        ),
+        mock.patch("shellflow.execute_remote", side_effect=_fake_execute_remote),
+    ):
+        result = run_script(blocks, servers, verbose=getattr(context, "verbose", False), mode=mode, variables=variables, macros=macros, helpers=helpers, hooks=hooks)
+
+    context.run_result = result
+    context.block_results = result.block_results
+    context.stdout = "\n".join(block.output for block in result.block_results if block.output)
+    context.stderr = result.error_message if not result.success else ""
+    context.exit_code = 0 if result.success else 1
+
+
 @when("the script is parsed")
 def step_when_the_script_is_parsed(context: Context) -> None:
     when_the_script_is_parsed(context)

src/advanced_modes.py

@@ -0,0 +1,110 @@
+"""Advanced execution modes for Shellflow.
+
+This module provides advanced execution modes including parallel execution
+and dry-run capabilities.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from typing import Any
+
+
+def run_parallel(
+    blocks,
+    context,
+    servers=None,
+    no_input=False,
+    verbose=False,
+    output_tail_lines=20,
+    run_id="",
+    total_blocks=0,
+):
+    """Execute blocks in parallel using a thread pool.
+
+    Args:
+        blocks: List of blocks to execute in parallel.
+        context: Shared execution context.
+        servers: SSH server configurations.
+        no_input: Whether to disable interactive input.
+        verbose: Whether to enable verbose output.
+        output_tail_lines: Maximum output lines to show.
+        run_id: Run identifier for events.
+        total_blocks: Total number of blocks in the script.
+
+    Returns:
+        List of execution results in the order they completed.
+    """
+    # Import here to avoid circular imports
+    from shellflow import (
+        ExecutionResult,
+        _execute_block_standard,
+        _finalize_block_result,
+        _apply_block_exports,
+        _get_verbose_colors,
+    )
+
+    if not blocks:
+        return []
+
+    # ANSI color codes for verbose output
+    colors = _get_verbose_colors()
+
+    results: list[ExecutionResult] = []
+    lock = threading.Lock()
+
+    def execute_block_worker(block: Block, block_index: int) -> ExecutionResult:
+        """Execute a single block and return its result."""
+        block_id = f"block-{block_index}"
+        start_time = time.perf_counter()
+
+        # Execute the block
+        result = _execute_block_standard(
+            block, context, servers, no_input, verbose, block_index, total_blocks,
+            output_tail_lines, colors, [], run_id
+        )
+
+        result = _finalize_block_result(result, block, block_index, start_time)
+        result.block_id = block_id
+        result.block_index = block_index
+
+        # Apply exports to shared context (thread-safe)
+        with lock:
+            result.exported_env = _apply_block_exports(block, result, context)
+            # Update shared context
+            context.last_output = result.output
+            context.success = result.success
+
+        return result
+
+    # Execute blocks in parallel
+    with ThreadPoolExecutor(max_workers=len(blocks)) as executor:
+        # Submit all tasks
+        future_to_block = {
+            executor.submit(execute_block_worker, block, i + 1): (block, i + 1)
+            for i, block in enumerate(blocks)
+        }
+
+        # Collect results as they complete
+        for future in as_completed(future_to_block):
+            block, block_index = future_to_block[future]
+            try:
+                result = future.result()
+                results.append(result)
+            except Exception as exc:
+                # Handle any exceptions from the worker thread
+                error_result = ExecutionResult(
+                    success=False,
+                    output="",
+                    error_message=f"Parallel execution failed: {exc}",
+                    block_index=block_index,
+                    source_line=block.source_line,
+                )
+                results.append(error_result)
+
+    # Sort results by block index to maintain order
+    results.sort(key=lambda r: r.block_index)
+
+    return results