# Platform Utilities This module provides cross-platform utilities to support Linux, macOS, and Windows:
Category Functions
Detection is_windows(), is_macos(), is_linux(), is_apple_silicon(), get_current_platform()
Paths get_python_in_env()
Process get_popen_isolation_kwargs(), terminate_process(), terminate_self()
Shell run_shell_command(), conda_env_exists()
Conda/Micromamba get_micromamba_download_url(), download_micromamba(), get_conda_command(), build_conda_command(), ensure_runtime_available()
## Platform Detection Functions to detect the current operating system and architecture. ------------------------------------------------------------------------ ### is_apple_silicon ``` python def is_apple_silicon( )->bool: ``` *Check if running on Apple Silicon Mac (for MPS detection).* ------------------------------------------------------------------------ ### is_linux ``` python def is_linux( )->bool: ``` *Check if running on Linux.* ------------------------------------------------------------------------ ### is_macos ``` python def is_macos( )->bool: ``` *Check if running on macOS.* ------------------------------------------------------------------------ ### is_windows ``` python def is_windows( )->bool: ``` *Check if running on Windows.* ``` python # Test detection functions print(f"is_windows(): {is_windows()}") print(f"is_macos(): {is_macos()}") print(f"is_linux(): {is_linux()}") print(f"is_apple_silicon(): {is_apple_silicon()}") # Exactly one of these should be True assert sum([is_windows(), is_macos(), is_linux()]) == 1 ``` is_windows(): False is_macos(): False is_linux(): True is_apple_silicon(): False ------------------------------------------------------------------------ ### get_current_platform ``` python def get_current_platform( )->str: ``` *Get current platform string for manifest filtering.* Returns strings like ‘linux-x64’, ‘darwin-arm64’, ‘win-x64’. ``` python # Test platform string current = get_current_platform() print(f"Current platform: {current}") # Should be one of the expected formats valid_prefixes = ["linux-", "darwin-", "win-"] assert any(current.startswith(p) for p in valid_prefixes) ``` Current platform: linux-x64 ## Path Utilities Functions for cross-platform path handling, particularly for conda environments. ------------------------------------------------------------------------ ### get_python_in_env ``` python def get_python_in_env( env_path:Path, # Path to conda environment root )->Path: # Path to Python executable ``` *Get the Python executable path for a conda environment.* On Windows: env_path/python.exe On Unix: env_path/bin/python ``` python # Test path generation test_env = Path("/home/user/miniforge3/envs/test-env") python_path = get_python_in_env(test_env) print(f"Python path: {python_path}") if is_windows(): assert python_path.name == "python.exe" else: assert "bin" in python_path.parts assert python_path.name == "python" ``` Python path: /home/user/miniforge3/envs/test-env/bin/python ## Process Management Cross-platform utilities for subprocess creation and termination. ------------------------------------------------------------------------ ### get_popen_isolation_kwargs ``` python def get_popen_isolation_kwargs( )->Dict: ``` *Return kwargs for process isolation in subprocess.Popen.* On Unix: Returns {‘start_new_session’: True} On Windows: Returns {‘creationflags’: CREATE_NEW_PROCESS_GROUP} Usage: process = subprocess.Popen(cmd, \*\*get_popen_isolation_kwargs(), …) ``` python # Test isolation kwargs kwargs = get_popen_isolation_kwargs() print(f"Isolation kwargs: {kwargs}") if is_windows(): assert "creationflags" in kwargs else: assert kwargs.get("start_new_session") == True ``` Isolation kwargs: {'start_new_session': True} ------------------------------------------------------------------------ ### terminate_process ``` python def terminate_process( process:Popen, # Process to terminate (must be a session/group leader for subtree kill) timeout:float=2.0, # Seconds to wait before force kill )->None: ``` *Terminate a subprocess + its entire process subtree (grandchildren, etc).* Session A 2026-05-27: enhanced from worker-only termination to FULL subtree termination. Workers are spawned with `get_popen_isolation_kwargs()` which sets `start_new_session=True` on Unix → the worker is its own session leader and ALL of its descendants inherit the same process-group ID (unless they setsid themselves, which is rare). `os.killpg(worker_pid, SIGTERM/SIGKILL)` sends the signal to every process in that group atomically — closes the orphan-grandchild bug surfaced by Voxtral-vLLM (vLLM api_server spawned its own EngineCore subprocess; pre-fix, the worker terminated cleanly but vLLM + EngineCore kept running as orphans, eating GPU memory until manual kill). Strategy on Unix: 1. SIGTERM the worker’s process group via os.killpg (atomic). 2. Wait up to `timeout` for the worker to exit. 3. If anything still alive, SIGKILL the process group. 4. psutil-based safety sweep for any process that setsid-ed away from the original group (rare but possible — e.g., a poorly-isolated subprocess). Strategy on Windows: 1. process.terminate() + wait + kill (legacy path). True process-group signaling on Windows requires Job Objects which the substrate doesn’t currently wire — Windows users are advised to avoid capabilities that spawn subprocesses until that’s added. (TODO: track as substrate gap.) ------------------------------------------------------------------------ ### terminate_self ``` python def terminate_self( )->None: ``` *Terminate the current process (for worker suicide pact).* On Unix: Sends SIGTERM to self for graceful shutdown On Windows: Calls os.\_exit() since Windows lacks SIGTERM ## Shell Command Execution Cross-platform shell command execution without hardcoded shell paths. ------------------------------------------------------------------------ ### run_shell_command ``` python def run_shell_command( cmd:str, # Shell command to execute check:bool=True, # Whether to raise on non-zero exit capture_output:bool=False, # Whether to capture stdout/stderr kwargs:VAR_KEYWORD )->CompletedProcess: # Additional kwargs passed to subprocess.run ``` *Run a shell command cross-platform.* Unlike using shell=True with executable=‘/bin/bash’, this function uses the platform’s default shell: - Linux/macOS: /bin/sh (or $SHELL) - Windows: cmd.exe ------------------------------------------------------------------------ ### conda_env_exists ``` python def conda_env_exists( env_name:str, # Name of the conda environment conda_cmd:str='conda', # Conda command (conda, mamba, micromamba) )->bool: ``` *Check if a conda environment exists (cross-platform).* Uses ‘conda env list –json’ instead of piping to grep, which doesn’t work on Windows. ``` python # Test shell command (simple echo) if is_windows(): result = run_shell_command("echo hello", capture_output=True) else: result = run_shell_command("echo hello", capture_output=True) print(f"Return code: {result.returncode}") print(f"Output: {result.stdout}") assert result.returncode == 0 ``` Running: echo hello Return code: 0 Output: b'hello\n' ``` python # Test conda_env_exists (will return False for non-existent env) exists = conda_env_exists("this-env-should-not-exist-12345") print(f"Non-existent env exists: {exists}") assert exists == False # Test with base env "miniforge3" (should exist if miniforge3 is installed) base_exists = conda_env_exists("miniforge3") print(f"Base env exists: {base_exists}") ``` Non-existent env exists: False Base env exists: True ## Conda/Micromamba Management Functions for managing conda and micromamba runtimes, including downloading micromamba binaries and building commands with proper prefix handling for project-local mode. ------------------------------------------------------------------------ ### get_micromamba_download_url ``` python def get_micromamba_download_url( platform_str:Optional=None, # Platform string (e.g., 'linux-x64'), uses current if None )->str: # Download URL for micromamba binary ``` *Get the micromamba download URL for the specified or current platform.* ------------------------------------------------------------------------ ### download_micromamba ``` python def download_micromamba( dest_path:Path, # Destination path for the micromamba binary platform_str:Optional=None, # Platform string, uses current if None show_progress:bool=True, # Whether to print progress messages )->bool: # True if download succeeded ``` *Download and extract micromamba binary to the specified path.* ``` python # Test micromamba URL functions current_platform = get_current_platform() url = get_micromamba_download_url() print(f"Platform: {current_platform}") print(f"Download URL: {url}") # Verify all platforms have URLs for plat in ["linux-x64", "linux-arm64", "darwin-x64", "darwin-arm64", "win-x64"]: assert plat in MICROMAMBA_URLS, f"Missing URL for {plat}" print(f" {plat}: {MICROMAMBA_URLS[plat]}") ``` Platform: linux-x64 Download URL: https://micro.mamba.pm/api/micromamba/linux-64/latest linux-x64: https://micro.mamba.pm/api/micromamba/linux-64/latest linux-arm64: https://micro.mamba.pm/api/micromamba/linux-aarch64/latest darwin-x64: https://micro.mamba.pm/api/micromamba/osx-64/latest darwin-arm64: https://micro.mamba.pm/api/micromamba/osx-arm64/latest win-x64: https://micro.mamba.pm/api/micromamba/win-64/latest ### Conda Command Building Functions to build conda/mamba/micromamba commands with proper prefix handling for project-local mode. ------------------------------------------------------------------------ ### get_conda_command ``` python def get_conda_command( config:CJMConfig, # Configuration object with runtime settings )->List: # Base command with prefix args if needed ``` *Get the conda/mamba/micromamba base command with prefix args for local mode.* ------------------------------------------------------------------------ ### build_conda_command ``` python def build_conda_command( config:CJMConfig, # Configuration object with runtime settings args:str, # Additional command arguments )->List: # Complete command ready for subprocess ``` *Build a complete conda/mamba/micromamba command.* ------------------------------------------------------------------------ ### get_micromamba_binary_path ``` python def get_micromamba_binary_path( config:CJMConfig, # Configuration object with runtime settings )->Optional: # Path to micromamba binary or None ``` *Get the configured micromamba binary path for the current platform.* ------------------------------------------------------------------------ ### ensure_runtime_available ``` python def ensure_runtime_available( config:CJMConfig, # Configuration object with runtime settings )->bool: # True if runtime is available ``` *Check if the configured conda/micromamba runtime is available.* ``` python # Test conda command building from cjm_substrate.core.config import CJMConfig, RuntimeConfig, CondaType, RuntimeMode # Test with default conda default_config = CJMConfig() cmd = get_conda_command(default_config) print(f"Default (conda): {cmd}") assert cmd == ["conda"] # Test with miniforge/mamba mamba_config = CJMConfig(runtime=RuntimeConfig(conda_type=CondaType.MINIFORGE)) cmd = get_conda_command(mamba_config) print(f"Miniforge (mamba): {cmd}") assert cmd == ["mamba"] # Test with micromamba system mode micromamba_system = CJMConfig(runtime=RuntimeConfig(conda_type=CondaType.MICROMAMBA)) cmd = get_conda_command(micromamba_system) print(f"Micromamba (system): {cmd}") assert cmd == ["micromamba"] # Test with micromamba local mode micromamba_local = CJMConfig( runtime=RuntimeConfig( conda_type=CondaType.MICROMAMBA, mode=RuntimeMode.LOCAL, prefix=Path("./runtime") ) ) cmd = get_conda_command(micromamba_local) print(f"Micromamba (local): {cmd}") # Path("./runtime") normalizes to "runtime" when converted to str assert cmd == ["micromamba", "-r", "runtime"] # Test build_conda_command full_cmd = build_conda_command(micromamba_local, "create", "-n", "test-env", "-y") print(f"Full command: {full_cmd}") assert full_cmd == ["micromamba", "-r", "runtime", "create", "-n", "test-env", "-y"] ``` Default (conda): ['conda'] Miniforge (mamba): ['mamba'] Micromamba (system): ['micromamba'] Micromamba (local): ['micromamba', '-r', 'runtime'] Full command: ['micromamba', '-r', 'runtime', 'create', '-n', 'test-env', '-y'] ## Summary This module provides the following cross-platform utilities: ### Detection - `is_windows()`, `is_macos()`, `is_linux()` - OS detection - `is_apple_silicon()` - Apple Silicon detection for MPS - `get_current_platform()` - Platform string like “linux-x64” ### Paths - `get_python_in_env(env_path)` - Python executable path in conda env ### Process Management - `get_popen_isolation_kwargs()` - Kwargs for subprocess isolation - `terminate_process(process, timeout)` - Graceful process termination - `terminate_self()` - Self-termination for suicide pact ### Shell Commands - `run_shell_command(cmd, ...)` - Cross-platform shell execution - `conda_env_exists(env_name)` - Check if conda env exists ### Conda/Micromamba Management - `MICROMAMBA_URLS` - Download URLs for micromamba binaries by platform - `get_micromamba_download_url(platform)` - Get download URL for current/specified platform - `download_micromamba(dest_path)` - Download and extract micromamba binary - `get_conda_command(config)` - Get base command with prefix args for local mode - `build_conda_command(config, *args)` - Build complete conda/micromamba command - `get_micromamba_binary_path(config)` - Get configured binary path for current platform - `ensure_runtime_available(config)` - Check if runtime is available