Method _dispatch_step

src/harness/execution/interpreter.py:258–362

Dispatch to the appropriate step handler based on step type. Args: step_data: Dictionary containing step configuration. context: Workflow context dictionary. step_number: Step identifier. args: EffectiveArgs with execution configuration.

(
        self,
        step_data: Dict[str, Any],
        context: Dict[str, Any],
        step_number: int,
        args,
        logger: Logger,
        checkpoint_manager: Optional[CheckpointManager] = None,
    )

            raise

    def _dispatch_step(
        self,
        step_data: Dict[str, Any],
        context: Dict[str, Any],
        step_number: int,
        args,
        logger: Logger,
        checkpoint_manager: Optional[CheckpointManager] = None,
    ):
        """Dispatch to the appropriate step handler based on step type.

        Args:
            step_data: Dictionary containing step configuration.
            context: Workflow context dictionary.
            step_number: Step identifier.
            args: EffectiveArgs with execution configuration.
            logger: Logger for output.
            checkpoint_manager: Optional checkpoint manager.

        Returns:
            Tuple of (output, tokens) from the step handler.
        """
        if "step" in step_data:
            return self.execute_basic_step(
                step_data, context, step_number, args, logger
            )
        elif "task" in step_data:
            return self.execute_task_step(step_data, context, step_number, args, logger)
        elif "for_each" in step_data:
            return self.execute_for_each_step(
                step_data,
                context,
                step_number,
                args,
                logger,
                checkpoint_manager=checkpoint_manager,
            )
        elif "if" in step_data:
            return self.execute_conditional_step(
                step_data,
                context,
                step_number,
                args,
                logger,
                checkpoint_manager=checkpoint_manager,
            )
        elif "parallel" in step_data:
            return self.execute_parallel_step(
                step_data,
                context,
                step_number,
                args,
                logger,
                checkpoint_manager=checkpoint_manager,
            )
        elif "while" in step_data:
            return self.execute_while_step(
                step_data,