PSPF 2025 bundle launcher that handles execution, extraction, and workenv setup.
PSPFLauncher(bundle_path: Path | None = None)
Bases: PSPFReader
Launch PSPF bundles.
Source code in flavor/psp/format_2025/launcher.py
| def __init__(self, bundle_path: Path | None = None) -> None:
if bundle_path is None:
# Allow None for testing purposes, parent class will handle it
bundle_path_arg: Path | str = ""
else:
bundle_path_arg = bundle_path
super().__init__(bundle_path_arg)
self.cache_dir = Path.home() / ".cache" / "flavor"
ensure_dir(self.cache_dir)
self._workenv_manager = WorkEnvManager(self)
|
acquire_lock(
lock_file: Path, timeout: float = 30.0
) -> Generator[Path, None, None]
Acquire a file-based lock for extraction.
Source code in flavor/psp/format_2025/launcher.py
| @contextmanager
def acquire_lock(self, lock_file: Path, timeout: float = 30.0) -> Generator[Path, None, None]:
"""Acquire a file-based lock for extraction."""
from flavor.locking import default_lock_manager
with default_lock_manager.lock(lock_file.name, timeout=timeout) as lock:
yield lock
|
check_disk_space(workenv_dir: Path) -> None
Check if there's enough disk space for extraction.
Parameters:
| Name |
Type |
Description |
Default |
workenv_dir
|
Path
|
Directory where slots will be extracted
|
required
|
Raises:
| Type |
Description |
OSError
|
If insufficient disk space available
|
Source code in flavor/psp/format_2025/launcher.py
| def check_disk_space(self, workenv_dir: Path) -> None:
"""Check if there's enough disk space for extraction.
Args:
workenv_dir: Directory where slots will be extracted
Raises:
OSError: If insufficient disk space available
"""
from provide.foundation.file import check_disk_space
# Calculate total size needed (compressed size * multiplier for safety)
slot_table = self.read_slot_table()
total_needed = sum(slot["size"] * DEFAULT_DISK_SPACE_MULTIPLIER for slot in slot_table)
# Use the utility function
check_disk_space(workenv_dir, total_needed)
|
execute(args: list[str] | None = None) -> dict[str, Any]
Execute the bundle.
Sets up the work environment, extracts slots, and executes the main command
using the BundleExecutor.
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, and other metadata
|
Source code in flavor/psp/format_2025/launcher.py
| def execute(self, args: list[str] | None = None) -> dict[str, Any]:
"""Execute the bundle.
Sets up the work environment, extracts slots, and executes the main command
using the BundleExecutor.
Args:
args: Command line arguments to pass to the executable
Returns:
dict: Execution result with exit_code, stdout, stderr, and other metadata
"""
try:
logger.info(f"🚀 Executing bundle: {self.bundle_path}")
# Read metadata
metadata = self.read_metadata()
# Validate execution configuration exists
if "execution" not in metadata:
logger.error("❌ No execution configuration in metadata")
raise ValueError("Bundle has no execution configuration")
# Setup work environment (extracts slots and runs setup commands)
workenv_dir = self.setup_workenv()
# Use the executor for actual process execution
from flavor.psp.format_2025.executor import BundleExecutor
logger.debug(f"🔍 Metadata command: {metadata.get('execution', {}).get('command', 'N/A')}")
logger.debug(f"🔍 Workenv dir: {workenv_dir}")
executor = BundleExecutor(metadata, workenv_dir)
# Execute and return result
return executor.execute(args)
except Exception as e:
logger.error(f"❌ Execution failed: {e}")
return {
"exit_code": 1,
"stdout": "",
"stderr": str(e),
"executed": False,
"command": None,
"args": args or [],
"pid": None,
"working_directory": str(Path.cwd()),
"error": str(e),
}
|
extract_all_slots(workenv_dir: Path) -> dict[int, Path]
Extract all slots to the work environment.
Parameters:
| Name |
Type |
Description |
Default |
workenv_dir
|
Path
|
Directory to extract slots into
|
required
|
Returns:
| Name | Type |
Description |
dict |
dict[int, Path]
|
Mapping of slot index to extracted path
|
Source code in flavor/psp/format_2025/launcher.py
| def extract_all_slots(self, workenv_dir: Path) -> dict[int, Path]:
"""Extract all slots to the work environment.
Args:
workenv_dir: Directory to extract slots into
Returns:
dict: Mapping of slot index to extracted path
"""
# NOTE: This parallels Go's ExtractAllSlots logic
slot_table = self.read_slot_table()
extracted_paths = {}
logger.info(f"📤 Extracting {len(slot_table)} slots")
try:
for slot_entry in slot_table:
slot_idx = slot_entry["index"]
logger.debug(f"🔄 Extracting slot {slot_idx}")
slot_path = self.extract_slot(slot_idx, workenv_dir)
extracted_paths[slot_idx] = slot_path
return extracted_paths
except Exception as e:
logger.error(f"❌ Extraction interrupted or failed: {e}. Cleaning up partial extraction.")
safe_rmtree(workenv_dir)
raise # Re-raise the exception
|
extract_slot(
slot_index: int,
workenv_dir: Path,
verify_checksum: bool = False,
) -> Path
Extract a single slot.
Parameters:
| Name |
Type |
Description |
Default |
slot_index
|
int
|
Index of the slot to extract
|
required
|
workenv_dir
|
Path
|
Directory to extract into
|
required
|
verify_checksum
|
bool
|
Whether to verify checksum after extraction
|
False
|
Returns:
| Name | Type |
Description |
Path |
Path
|
Path to the extracted slot content
|
Source code in flavor/psp/format_2025/launcher.py
| def extract_slot(self, slot_index: int, workenv_dir: Path, verify_checksum: bool = False) -> Path:
"""Extract a single slot.
Args:
slot_index: Index of the slot to extract
workenv_dir: Directory to extract into
verify_checksum: Whether to verify checksum after extraction
Returns:
Path: Path to the extracted slot content
"""
slot_table = self.read_slot_table()
if slot_index < 0 or slot_index >= len(slot_table):
logger.error(f"❌ Invalid slot index: {slot_index} (have {len(slot_table)} slots)")
raise ValueError(f"Invalid slot index: {slot_index}")
slot_entry = slot_table[slot_index]
logger.debug(
f"📍 Slot {slot_index}: offset={slot_entry['offset']}, size={slot_entry['size']}, operations={slot_entry['operations']}"
)
# Read slot data from bundle
with Path(self.bundle_path).open("rb") as f:
f.seek(slot_entry["offset"])
slot_data = f.read(slot_entry["size"])
if verify_checksum:
self._verify_slot_checksum(slot_data, slot_entry, slot_index)
data = self._decode_slot_data(slot_data, slot_entry["operations"], slot_index)
slot_name = self._get_slot_name(slot_index)
logger.debug(f"📝 Slot {slot_index} name: {slot_name}")
# Check if it's a tarball that needs extraction
if self._is_tarball(data) or slot_name.endswith((".tar.gz", ".tgz")):
logger.debug(f"📤 Extracting tarball {slot_name} to {workenv_dir}")
try:
with tarfile.open(fileobj=io.BytesIO(data), mode="r:*") as tar:
tar.extractall(path=workenv_dir, filter="data")
return workenv_dir
except (OSError, PermissionError, tarfile.ReadError) as e:
logger.error(f"❌ Disk or tarball error extracting slot {slot_index} to {workenv_dir}: {e}")
raise
# Write single file (atomic for safety)
output_path = workenv_dir / slot_name
try:
ensure_parent_dir(output_path)
atomic_write(output_path, data)
return output_path
except (OSError, PermissionError) as e:
logger.error(f"❌ Disk error writing slot {slot_index} to {output_path}: {e}")
raise
|
read_slot_table() -> list[dict[str, Any]]
Read the slot table from the bundle.
Returns:
| Name | Type |
Description |
list |
list[dict[str, Any]]
|
List of slot entries, each containing:
- offset: Start position of slot data
- size: Size of uncompressed data
- checksum: Adler32 checksum
- encoding: 0=none, 1=gzip, 2=reserved
- purpose: 0=payload, 1=runtime, 2=tool
- lifecycle: 0=persistent, 1=volatile, 2=temporary, 3=install
|
Source code in flavor/psp/format_2025/launcher.py
| def read_slot_table(self) -> list[dict[str, Any]]:
"""Read the slot table from the bundle.
Returns:
list: List of slot entries, each containing:
- offset: Start position of slot data
- size: Size of uncompressed data
- checksum: Adler32 checksum
- encoding: 0=none, 1=gzip, 2=reserved
- purpose: 0=payload, 1=runtime, 2=tool
- lifecycle: 0=persistent, 1=volatile, 2=temporary, 3=install
"""
# NOTE: This logic is unique to Python launcher - Go/Rust have their own implementations
index = self.read_index()
slot_entries = []
with Path(self.bundle_path).open("rb") as f:
# Seek to slot table
f.seek(index.slot_table_offset)
# Read each 64-byte slot descriptor (new format)
for i in range(index.slot_count):
entry_data = f.read(DEFAULT_SLOT_DESCRIPTOR_SIZE)
if len(entry_data) != DEFAULT_SLOT_DESCRIPTOR_SIZE:
raise ValueError(
f"Invalid slot table entry {i}: expected {DEFAULT_SLOT_DESCRIPTOR_SIZE} bytes, got {len(entry_data)}"
)
# Use SlotDescriptor to unpack
from flavor.psp.format_2025.slots import SlotDescriptor
descriptor = SlotDescriptor.unpack(entry_data)
# Extract the fields we need for launcher
offset = descriptor.offset
size = descriptor.size # Compressed size
checksum = descriptor.checksum
operations = descriptor.operations
purpose = descriptor.purpose
lifecycle = descriptor.lifecycle
slot_entries.append(
{
"index": i,
"offset": offset,
"size": size,
"checksum": checksum,
"operations": operations,
"purpose": purpose,
"lifecycle": lifecycle,
}
)
return slot_entries
|
Setup work environment for bundle execution.
Source code in flavor/psp/format_2025/launcher.py
| def setup_workenv(self) -> Path:
"""Setup work environment for bundle execution."""
return self._workenv_manager.setup_workenv(self.bundle_path)
|
verify_integrity() -> dict[str, bool]
Verify package integrity including signatures and checksums.
Returns:
Source code in flavor/psp/format_2025/launcher.py
| def verify_integrity(self) -> dict[str, bool]:
"""
Verify package integrity including signatures and checksums.
Returns:
Dictionary with verification results:
- valid: Overall validity
- signature_valid: Signature verification result
- tamper_detected: Whether tampering was detected
"""
from flavor.psp.protocols import IntegrityResult
from flavor.psp.security import verify_package_integrity
if not self.bundle_path:
return {"valid": False, "signature_valid": False, "tamper_detected": True}
result: IntegrityResult = verify_package_integrity(self.bundle_path)
# IntegrityResult is a TypedDict with bool values, which is compatible with dict[str, bool]
return dict(result) # type: ignore[arg-type]
|