Method execute_workflow_step

src/harness/execution/interpreter.py:132–256

Execute a single workflow step based on its type. This is the main entry point for step execution. It logs step start/end events and dispatches to the appropriate handler. Args: step_data: Dictionary containing step configuration. context: Workflow c

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

        return load_submodule_tools_for_yaml(yaml_data, yaml_file_path)

    def execute_workflow_step(
        self,
        step_data: Dict[str, Any],
        context: Dict[str, Any],
        step_number: int,
        args,
        logger: Logger,
        checkpoint_manager: Optional[CheckpointManager] = None,
    ):
        """Execute a single workflow step based on its type.

        This is the main entry point for step execution. It logs step
        start/end events and dispatches to the appropriate handler.

        Args:
            step_data: Dictionary containing step configuration.
            context: Workflow context dictionary.
            step_number: Step identifier (can be hierarchical like "1.2.3").
            args: EffectiveArgs with execution configuration.
            logger: Logger for output and event logging.
            checkpoint_manager: Optional checkpoint manager for durability.

        Returns:
            Tuple of (output, tokens) where output is the step result
            and tokens is the number of tokens used.

        Raises:
            Exception: Any exception from step execution is re-raised
                      after logging the error.
        """
        if not isinstance(step_data, dict):
            logger(f"Unknown step type: {step_data}")
            step_id = str(step_number)
            logger.log_event(
                logger,
                "step_start",
                {
                    "step_id": step_id,
                    "step_number": step_id,
                    "name": f"unknown {step_id}",
                    "step_type": "unknown",
                    "instruction": "",
                    "depth": get_step_depth(step_id),
                },
            )
            logger.log_event(
                logger,
                "step_end",
                {
                    "step_id": step_id,
                    "tokens": 0,
                    "status": "error",
                    "error": "Invalid step format (expected mapping)",
                },
            )
            return "", 0

        step_id = str(step_number)