workenv

🤖 AI-Generated Content

This documentation was generated with AI assistance and is still being audited. Some, or potentially a lot, of this information may be inaccurate. Learn more.

flavor.psp.format_2025.workenv

PSPF Work Environment Management

Handles work environment setup, caching, lifecycle management, and setup commands.

Classes

WorkEnvManager

WorkEnvManager(reader: PSPFReader)

Manages PSPF work environments.

Initialize with reference to PSPFReader.

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

def __init__(self, reader: PSPFReader) -> None:
    """Initialize with reference to PSPFReader."""
    self.reader = reader

Functions

setup_workenv

setup_workenv(bundle_path: Path) -> Path

Setup work environment for bundle execution.

Creates a work environment directory, extracts slots, and runs setup commands. Uses cache validation to avoid re-extraction when possible. Handles lifecycle-based slot cleanup (e.g., 'init' slots removed after setup).

Parameters:

Name	Type	Description	Default
bundle_path	Path	Path to the bundle	required

Returns:

Name	Type	Description
Path	Path	Path to the work environment directory

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

def setup_workenv(self, bundle_path: Path) -> Path:
    """Setup work environment for bundle execution.

    Creates a work environment directory, extracts slots, and runs setup commands.
    Uses cache validation to avoid re-extraction when possible.
    Handles lifecycle-based slot cleanup (e.g., 'init' slots removed after setup).

    Args:
        bundle_path: Path to the bundle

    Returns:
        Path: Path to the work environment directory
    """

    # NOTE: This matches Go's work environment setup logic
    metadata = self.reader.read_metadata()
    package_name = metadata["package"]["name"]
    package_version = metadata["package"]["version"]

    # Create work environment directory
    workenv_base = Path.home() / ".cache" / "flavor" / "workenv"
    workenv_dir = workenv_base / f"{package_name}_{package_version}"
    ensure_dir(workenv_dir)

    # Check cache validity
    cache_valid = self._check_cache_validity(metadata, workenv_dir, package_version)

    # Extract slots if cache is invalid
    if not cache_valid:
        logger.info("📤 Extracting slots (cache invalid)")
        # Extract all slots by iterating through slot count
        extracted_slots: dict[int, Path] = {}
        assert self.reader._index is not None
        slot_count = self.reader._index.slot_count
        for slot_idx in range(slot_count):
            slot_path = self.reader.extract_slot(slot_idx, workenv_dir)
            extracted_slots[slot_idx] = slot_path

        # Run setup commands
        if "setup_commands" in metadata:
            self._run_setup_commands(metadata["setup_commands"], workenv_dir, metadata)

        # Handle lifecycle-based cleanup
        self._cleanup_lifecycle_slots(workenv_dir, metadata, extracted_slots)
    else:
        pass

    return workenv_dir

substitute_slot_references

substitute_slot_references(
    command: str, workenv_dir: Path
) -> str

Substitute {slot:N} references in command.

Parameters:

Name	Type	Description	Default
command	str	Command with potential slot references	required
workenv_dir	Path	Work environment directory	required

Returns:

Name	Type	Description
str	str	Command with slot references substituted

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

def substitute_slot_references(self, command: str, workenv_dir: Path) -> str:
    """Substitute {slot:N} references in command.

    Args:
        command: Command with potential slot references
        workenv_dir: Work environment directory

    Returns:
        str: Command with slot references substituted
    """
    # NOTE: Slot substitution logic matches Go implementation
    metadata = self.reader.read_metadata()

    for i, slot in enumerate(metadata.get("slots", [])):
        placeholder = f"{{slot:{i}}}"
        if placeholder in command:
            slot_name = slot.get("id", f"slot_{i}")
            slot_path = workenv_dir / slot_name
            command = command.replace(placeholder, str(slot_path))
            logger.debug(f"🔄 Substituted {placeholder} -> {slot_path}")

    return command

Functions