executor

flavor.psp.format_2025.executor

PSPF 2025 Bundle Executor Handles process execution with environment setup and variable substitution.

Classes

BundleExecutor

BundleExecutor(metadata: dict[str, Any], workenv_dir: Path)

Executes PSPF bundles with proper environment and substitution.

Initialize executor with metadata and work environment.

Parameters:

Name	Type	Description	Default
metadata	dict[str, Any]	Bundle metadata containing execution configuration	required
workenv_dir	Path	Path to the extracted work environment	required

Source code in flavor/psp/format_2025/executor.py

def __init__(self, metadata: dict[str, Any], workenv_dir: Path) -> None:
    """Initialize executor with metadata and work environment.

    Args:
        metadata: Bundle metadata containing execution configuration
        workenv_dir: Path to the extracted work environment
    """
    self.metadata = metadata
    self.workenv_dir = workenv_dir
    self.package_name = metadata.get("package", {}).get("name", "unknown")
    self.package_version = metadata.get("package", {}).get("version", "")
    self.execution_config = metadata.get("execution", {})

Functions

execute

execute(args: list[str] | None = None) -> dict[str, Any]

Execute the bundle command.

Parameters:

Name	Type	Description	Default
args	list[str] | None	Command line arguments to pass to the executable	None

Returns:

Name	Type	Description
dict	dict[str, Any]	Execution result with exit_code, stdout, stderr, etc.

Source code in flavor/psp/format_2025/executor.py

def execute(self, args: list[str] | None = None) -> dict[str, Any]:
    """Execute the bundle command.

    Args:
        args: Command line arguments to pass to the executable

    Returns:
        dict: Execution result with exit_code, stdout, stderr, etc.
    """
    # Get base command
    command = self.execution_config.get("command", "")
    if not command:
        raise ValueError("No command specified in execution configuration")

    # Prepare command with substitutions
    command = self.prepare_command(command, args)

    # Prepare environment
    env = self.prepare_environment()

    logger.info(f"🏃 Executing: {command}")

    try:
        # Parse command into arguments (safely handles quotes and spaces)
        command_args = shlex.split(command)

        # Execute the command using shared utility (no shell=True for security)
        result = run(
            command_args,
            cwd=self.workenv_dir,
            env=env,
            capture_output=True,
            check=False,  # We want to handle the exit code ourselves
        )

        # Log result
        if result.returncode == 0:
            pass
        else:
            logger.warning(f"⚠️ Execution completed with exit code: {result.returncode}")
            if result.stderr:
                logger.debug(f"📝 stderr: {result.stderr[:500]}")  # Log first 500 chars

        crashed = result.returncode < 0  # Negative return codes often indicate a crash due to a signal
        return {
            "exit_code": result.returncode,
            "stdout": result.stdout,
            "stderr": result.stderr,
            "executed": True,
            "command": command,
            "args": args or [],  # Return the original user args, not the parsed command
            "pid": os.getpid(),  # Current process PID since we don't have access to subprocess PID
            "working_directory": str(self.workenv_dir),
            "error": None if result.returncode == 0 else f"Process exited with code {result.returncode}",
            "crashed": crashed,
        }

    except Exception as e:
        logger.error(f"❌ Execution failed: {e}")
        return {
            "exit_code": 1,
            "stdout": "",
            "stderr": str(e),
            "executed": False,
            "command": command,
            "args": args or [],  # Return the original user args
            "pid": None,
            "working_directory": str(self.workenv_dir),
            "error": str(e),
            "returncode": 1,  # Add returncode for consistency
        }

prepare_command

prepare_command(
    base_command: str, args: list[str] | None = None
) -> str

Prepare command with substitutions and arguments.

Parameters:

Name	Type	Description	Default
base_command	str	Command template with placeholders	required
args	list[str] | None	Additional arguments to append	None

Returns:

Name	Type	Description
str	str	Prepared command ready for execution

Source code in flavor/psp/format_2025/executor.py

def prepare_command(self, base_command: str, args: list[str] | None = None) -> str:
    """Prepare command with substitutions and arguments.

    Args:
        base_command: Command template with placeholders
        args: Additional arguments to append

    Returns:
        str: Prepared command ready for execution
    """
    logger.debug(f"🔍 prepare_command input: {base_command}")

    # Primary slot substitution
    command = self._substitute_primary(base_command)
    logger.debug(f"🔍 after primary substitution: {command}")

    # Slot substitution - {slot:N} references
    command = self._substitute_slots(command)
    logger.debug(f"🔍 after slot substitution: {command}")

    # Basic substitutions - only {workenv}, {package_name}, and {version} as per spec
    command = command.replace("{workenv}", str(self.workenv_dir))
    command = command.replace("{package_name}", self.package_name)
    command = command.replace("{version}", self.package_version)
    logger.debug(f"🔍 after basic substitutions: {command}")

    # Append user arguments
    if args:
        arg_str = " ".join(f'"{arg}"' if " " in arg else arg for arg in args)
        command = f"{command} {arg_str}"

    return command

prepare_environment

prepare_environment() -> dict[str, str]

Prepare environment variables for execution.

Returns:

Name	Type	Description
dict	dict[str, str]	Environment variables including FLAVOR_* vars

Source code in flavor/psp/format_2025/executor.py

def prepare_environment(self) -> dict[str, str]:
    """Prepare environment variables for execution.

    Returns:
        dict: Environment variables including FLAVOR_* vars
    """
    env = os.environ.copy()

    # Standard FLAVOR environment variables
    env["FLAVOR_WORKENV"] = str(self.workenv_dir)
    env["FLAVOR_PACKAGE"] = self.package_name
    env["FLAVOR_VERSION"] = self.package_version

    # Custom environment variables from metadata
    if "env" in self.execution_config:
        for key, value in self.execution_config["env"].items():
            value = str(value).replace("{workenv}", str(self.workenv_dir))
            value = value.replace("{package_name}", self.package_name)
            value = value.replace("{version}", self.package_version)
            env[key] = value

    return env