Platform Utilities

Cross-platform utilities for process management, path handling, and system detection

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

def is_apple_silicon(
    
)->bool:

Check if running on Apple Silicon Mac (for MPS detection).

---

is_linux

def is_linux(
    
)->bool:

Check if running on Linux.

---

is_macos

def is_macos(
    
)->bool:

Check if running on macOS.

---

is_windows

def is_windows(
    
)->bool:

Check if running on Windows.

# 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

def get_current_platform(
    
)->str:

Get current platform string for manifest filtering.

Returns strings like ‘linux-x64’, ‘darwin-arm64’, ‘win-x64’.

# 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

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

# 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

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(), …)

# 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

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 plugins that spawn subprocesses until that’s added. (TODO: track as substrate gap.)

---

terminate_self

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

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

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.

# 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'

# 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

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

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.

# 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

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

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

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

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.

# Test conda command building
from cjm_plugin_system.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