Class YAMLTaskParser

src/harness/parsing/yaml_parser.py:15–280

YAML task parser

class YAMLTaskParser:
    """YAML task parser"""

    def __init__(self):
        self.context = {}
        self.variables = {}

    def load_task(self, file_path: str) -> Dict[str, Any]:
        """Load and parse a YAML task file"""
        with open(file_path, "r", encoding="utf-8") as f:
            content = f.read()

        # Expand environment variables first
        content = self._expand_env_variables(content)

        # Parse YAML
        task_data = yaml.safe_load(content)

        # Validate task format
        self._validate_task(task_data)

        return task_data

    def _expand_env_variables(self, content: str) -> str:
        """Expand environment variables and template variables"""

        # Handle ${VAR} and ${VAR:-default} formats
        def replace_var(match):
            var_expr = match.group(1)
            if ":-" in var_expr:
                var_name, default_value = var_expr.split(":-", 1)
                value = os.getenv(var_name, default_value)
            else:
                value = os.getenv(var_expr, match.group(0))

            # If the value contains special YAML characters and needs escaping
            if value and value != match.group(
                0
            ):  # Only if we actually got a value from env
                # Escape backslashes and double quotes for YAML double-quoted strings
                # This handles JSON strings and other special content
                if (
                    "\\" in value
                    or '"' in value
                    or "\n" in value
                    or "\r" in value
                    or "\t" in value
                ):
                    value = value.replace("\\", "\\\\")  # Escape backslashes first
                    value = value.replace('"', '\\"')  # Escape double quotes
                    value = value.replace("\n", "\\n")  # Escape newlines
                    value = value.replace("\r", "\\r")  # Escape carriage returns
                    value = value.replace("\t", "\\t")  # Escape tabs

            return value

        # Replace environment variables
        content = re.sub(r"\$\{([^}]+)\}", replace_var, content)