Core API

Core modules providing Axium's foundational functionality.

---

env

Environment management and configuration.

env

Environment abstraction layer for Axium.

Provides access to environment configuration from envs.yaml and tracks the active environment in state.json. All environment properties are accessible via template expansion ${env.}.

Example envs.yaml:

envs:
  root:
    prefix: enva-root
    aws_profile: root
    color: teal
    region: eu-west-1
  builder:
    prefix: enva-builder
    aws_profile: builder
    color: cyan
    region: eu-west-2

Example state.json:

{"active_env": "root"}

Usage

>>> import axium.core.env as env
>>> env.get_active_env_name()
'root'
>>> env.get_env_value('prefix')
'enva-root'
>>> env.get_env_value('region')
'eu-west-1'

load_envs()

Load all environments from envs.yaml.

Returns:

Type	Description
dict[str, dict[str, Any]]	Dictionary mapping environment names to their properties.
dict[str, dict[str, Any]]	Returns empty dict if file doesn't exist or is invalid.

Example

>>> load_envs()
{'root': {'prefix': 'enva-root', 'region': 'eu-west-1'}, ...}

Source code in axium/core/env.py

def load_envs() -> dict[str, dict[str, Any]]:
    """
    Load all environments from envs.yaml.

    Returns:
        Dictionary mapping environment names to their properties.
        Returns empty dict if file doesn't exist or is invalid.

    Example:
        ```python
        >>> load_envs()
        {'root': {'prefix': 'enva-root', 'region': 'eu-west-1'}, ...}
        ```
    """
    if not ENVS_PATH.exists():
        logger.debug("No envs.yaml found at %s", ENVS_PATH)
        return {}

    try:
        data = yaml.safe_load(ENVS_PATH.read_text())
        if not data or "envs" not in data:
            logger.warning("envs.yaml missing 'envs' key")
            return {}

        envs = data["envs"]
        if not isinstance(envs, dict):
            logger.warning("envs.yaml 'envs' is not a dictionary")
            return {}

        logger.debug("Loaded %d environments from envs.yaml", len(envs))
        return envs

    except Exception as e:
        logger.error("Failed to load envs.yaml: %s", e)
        return {}

get_active_env_name()

Get the name of the currently active environment from state.json.

Returns:

Type	Description
str | None	Active environment name, or None if not set or file missing.

Example

>>> get_active_env_name()
'root'

Source code in axium/core/env.py

def get_active_env_name() -> str | None:
    """
    Get the name of the currently active environment from state.json.

    Returns:
        Active environment name, or None if not set or file missing.

    Example:
        ```python
        >>> get_active_env_name()
        'root'
        ```
    """
    if not STATE_PATH.exists():
        logger.debug("No state.json found at %s", STATE_PATH)
        return None

    try:
        data = json.loads(STATE_PATH.read_text())
        active_env = data.get("active_env")
        logger.debug("Active environment: %s", active_env)
        return active_env

    except Exception as e:
        logger.error("Failed to load state.json: %s", e)
        return None

get_active_env()

Get the full configuration dictionary for the active environment.

Returns:

Type	Description
dict[str, Any]	Dictionary of environment properties, or empty dict if no active env
dict[str, Any]	or environment not found in envs.yaml.

Example

>>> get_active_env()
{'prefix': 'enva-root', 'aws_profile': 'root', 'region': 'eu-west-1'}

Source code in axium/core/env.py

def get_active_env() -> dict[str, Any]:
    """
    Get the full configuration dictionary for the active environment.

    Returns:
        Dictionary of environment properties, or empty dict if no active env
        or environment not found in envs.yaml.

    Example:
        ```python
        >>> get_active_env()
        {'prefix': 'enva-root', 'aws_profile': 'root', 'region': 'eu-west-1'}
        ```
    """
    env_name = get_active_env_name()
    if not env_name:
        logger.debug("No active environment set")
        return {}

    envs = load_envs()
    env_data = envs.get(env_name, {})

    if not env_data:
        logger.warning("Active environment '%s' not found in envs.yaml", env_name)

    return env_data

get_env_value(key)

Get a specific property value from the active environment.

Parameters:

Name	Type	Description	Default
key	str	Property name (e.g., "prefix", "region", "aws_profile")	required

Returns:

Type	Description
Any	Property value, or None if key doesn't exist or no active env.

Example

>>> get_env_value('prefix')
'enva-root'
>>> get_env_value('region')
'eu-west-1'
>>> get_env_value('nonexistent')
None

Source code in axium/core/env.py

def get_env_value(key: str) -> Any:
    """
    Get a specific property value from the active environment.

    Args:
        key: Property name (e.g., "prefix", "region", "aws_profile")

    Returns:
        Property value, or None if key doesn't exist or no active env.

    Example:
        ```python
        >>> get_env_value('prefix')
        'enva-root'
        >>> get_env_value('region')
        'eu-west-1'
        >>> get_env_value('nonexistent')
        None
        ```
    """
    env_data = get_active_env()
    value = env_data.get(key)

    if value is None:
        logger.debug("Key '%s' not found in active environment", key)

    return value

validate_env_name(name)

Validate that an environment name exists in envs.yaml.

Parameters:

Name	Type	Description	Default
name	str | None	Environment name to validate	required

Returns:

Type	Description
bool	Tuple of (is_valid, error_message).
str | None	If valid: (True, None)
tuple[bool, str | None]	If invalid: (False, "error message with suggestions")

Example

>>> validate_env_name("root")
(True, None)
>>> validate_env_name("invalid")
(False, "Unknown environment: invalid\nDid you mean: root, prod?")
>>> validate_env_name(None)
(False, "Environment name cannot be empty")

Source code in axium/core/env.py

def validate_env_name(name: str | None) -> tuple[bool, str | None]:
    """
    Validate that an environment name exists in envs.yaml.

    Args:
        name: Environment name to validate

    Returns:
        Tuple of (is_valid, error_message).
        If valid: (True, None)
        If invalid: (False, "error message with suggestions")

    Example:
        ```python
        >>> validate_env_name("root")
        (True, None)
        >>> validate_env_name("invalid")
        (False, "Unknown environment: invalid\\nDid you mean: root, prod?")
        >>> validate_env_name(None)
        (False, "Environment name cannot be empty")
        ```
    """
    if not name:
        return False, "Environment name cannot be empty"

    envs = load_envs()

    if not envs:
        # No envs.yaml file - warn but allow (for initial setup)
        logger.warning("No environments configured in envs.yaml")
        return True, None

    if name in envs:
        return True, None

    # Environment not found - build error with suggestions
    import difflib

    suggestions = difflib.get_close_matches(name, envs.keys(), n=3, cutoff=0.6)

    if suggestions:
        error = f"Unknown environment: {name}\nDid you mean: {', '.join(suggestions)}?"
    else:
        error = f"Unknown environment: {name}"

    return False, error

---

prefix

Command prefix system and template expansion.

prefix

Prefix system for command wrapping.

This module handles loading prefix rules from configuration and applying them to commands based on context (environment, pane, etc.).

PrefixRegistry

Registry for command prefix rules.

Tracks rules from both prefixes.yaml and programmatic registration (from spokes/gears). Provides conflict detection to prevent multiple components from registering the same command.

Source code in axium/core/prefix.py

class PrefixRegistry:
    """
    Registry for command prefix rules.

    Tracks rules from both prefixes.yaml and programmatic registration
    (from spokes/gears). Provides conflict detection to prevent multiple
    components from registering the same command.
    """

    def __init__(self):
        self.rules: list[dict[str, Any]] = []
        self.owners: dict[str, str] = {}  # {command: owner_name} for conflict detection
        self._load_yaml_rules()

    def _load_yaml_rules(self) -> None:
        """Load prefix rules from prefixes.yaml."""
        if not PREFIXES_PATH.exists():
            logger.debug("No prefixes.yaml found at %s", PREFIXES_PATH)
            return

        try:
            data = yaml.safe_load(PREFIXES_PATH.read_text())
            rules = data.get("prefixes", [])
            for rule in rules:
                if "command" in rule and "prefix" in rule:
                    # Mark yaml rules as owned by "config"
                    rule_with_owner = rule.copy()
                    if "owner" not in rule_with_owner:
                        rule_with_owner["owner"] = "config"
                    self.rules.append(rule_with_owner)
                    self.owners[rule["command"]] = rule_with_owner["owner"]
            logger.info("Loaded %d prefix rules from prefixes.yaml", len(self.rules))
        except Exception as e:
            logger.error("Failed to load prefixes.yaml: %s", e)

    def register_rule(self, command: str, wrapper: str, owner: str) -> bool:
        """
        Register a prefix rule programmatically (for Spokes/Gears).

        Args:
            command: Command to intercept (e.g., "ansible-playbook")
            wrapper: Wrapper command (e.g., "axium ansible-run")
            owner: Name of spoke/gear registering this rule

        Returns:
            True if registered successfully, False if conflict detected

        Example:
            ```python
            >>> registry = PrefixRegistry()
            >>> registry.register_rule("ansible-playbook", "axium ansible-run", "ansible")
            True
            >>> registry.register_rule("ansible-playbook", "other command", "terraform")
            False  # Conflict!
            ```

        Note:
            Logs warning and returns False on conflict.
            Use get_rule_owner() to check current owner before registering.
        """
        # Check for conflict
        if command in self.owners:
            existing_owner = self.owners[command]
            if existing_owner != owner:
                logger.warning(
                    "Prefix conflict: '%s' tried to register command '%s' but already owned by '%s'",
                    owner,
                    command,
                    existing_owner,
                )
                return False
            else:
                # Same owner re-registering (e.g., during reload) - allow and update
                logger.debug(
                    "Updating prefix rule for command '%s' (owner: %s)", command, owner
                )
                # Remove old rule
                self.rules = [
                    r
                    for r in self.rules
                    if r.get("command") != command or r.get("owner") != owner
                ]

        # Register new rule
        rule = {
            "command": command,
            "prefix": wrapper,  # Note: using 'prefix' key for consistency with yaml format
            "owner": owner,
        }
        self.rules.append(rule)
        self.owners[command] = owner
        logger.info(
            "Registered prefix rule: %s → %s (owner: %s)", command, wrapper, owner
        )
        return True

    def unregister_rules_for_owner(self, owner: str) -> int:
        """
        Unregister all prefix rules for a specific owner (spoke/gear).

        Used when unloading a spoke/gear to clean up its registered rules.

        Args:
            owner: Name of spoke/gear to remove rules for

        Returns:
            Number of rules removed

        Example:
            ```python
            >>> registry.unregister_rules_for_owner("ansible")
            2  # Removed 2 rules
            ```
        """
        removed_commands = [
            cmd for cmd, cmd_owner in self.owners.items() if cmd_owner == owner
        ]
        self.rules = [r for r in self.rules if r.get("owner") != owner]

        for cmd in removed_commands:
            del self.owners[cmd]

        if removed_commands:
            logger.info(
                "Unregistered %d prefix rules for owner: %s",
                len(removed_commands),
                owner,
            )

        return len(removed_commands)

    def get_rule_owner(self, command: str) -> str | None:
        """
        Get the owner (spoke/gear name) of a prefix rule.

        Args:
            command: Command to check

        Returns:
            Owner name or None if no rule registered

        Example:
            ```python
            >>> registry.get_rule_owner("ansible-playbook")
            'ansible'
            ```
        """
        return self.owners.get(command)

    def get_all_rules(self) -> list[dict[str, Any]]:
        """
        Get all registered prefix rules with metadata.

        Returns:
            List of rule dicts with command, prefix/wrapper, and owner

        Example:
            ```python
            >>> rules = registry.get_all_rules()
            >>> for rule in rules:
            ...     print(f"{rule['command']} owned by {rule['owner']}")
            ```
        """
        return self.rules.copy()

    def get_prefixed_commands(self) -> list[str]:
        """Get list of all commands that have prefix rules."""
        commands = set()
        for rule in self.rules:
            commands.add(rule["command"])
        return sorted(commands)

    def apply_prefixes(
        self, command: str, args: list[str], context: dict[str, Any]
    ) -> tuple[list[str], dict[str, str]]:
        """
        Apply prefix rules to a command.

        Args:
            command: The command to prefix (e.g., "aws")
            args: Command arguments (e.g., ["s3", "ls"])
            context: Context dict with 'env', 'pane', etc.

        Returns:
            Tuple of (command_array, env_vars)
            command_array: Full command to execute
                (e.g., ["enva-root", "aws", "s3", "ls"])
            env_vars: Environment variables to set
                (currently unused, for future)
        """
        # Find matching rules for this command
        matching_rules = [r for r in self.rules if r["command"] == command]

        if not matching_rules:
            # No rules, return unwrapped command
            return ([command] + args, {})

        # Apply first matching rule (TODO: support multiple rules/conditions)
        rule = matching_rules[0]
        prefix_template = rule["prefix"]

        # Expand template variables
        expanded_prefix = self._expand_template(prefix_template, context)

        if not expanded_prefix:
            # Template expansion failed (e.g., missing variable)
            logger.warning("Failed to expand prefix template: %s", prefix_template)
            return ([command] + args, {})

        # Return prefixed command
        final_command = [expanded_prefix, command] + args
        logger.debug("Prefixed command: %s", " ".join(final_command))
        return (final_command, {})

    def _expand_template(self, template: str, context: dict[str, Any]) -> str:
        """
        Expand template variables.

        Supports two types of variables:
        1. ${env.<key>} - Environment properties from envs.yaml
        2. ${var} - Context variables (tmux_pane, etc.)

        Args:
            template: Template string
                (e.g., "${env.prefix}" or "pane-${tmux_pane}")
            context: Context dict with variable values
                (must include 'env' key with environment name)

        Returns:
            Expanded string, or empty string if required variables missing
        """
        # First, expand ${env.<key>} variables from envs.yaml
        # Load environment data from context if available
        env_dict = None
        env_name = context.get("env")
        if env_name:
            # Load the specific environment's data
            envs = env.load_envs()
            env_dict = envs.get(env_name, {})

        # Use expand_template directly with the loaded env_dict
        from axium.core.utils import expand_template

        result = expand_template(
            template,
            env_dict=env_dict,
            expand_tilde=False,
            expand_env_vars=False,
            expand_env_keys=True,
        )

        # If expansion failed (returned original and had env vars), fail
        if result == template and re.search(r"\$\{env\.", template):
            # Had ${env.} variables but they didn't expand
            return ""

        # Now expand context variables like ${tmux_pane}
        pattern = r"\$\{([^}]+)\}"
        matches = re.findall(pattern, result)

        if not matches:
            # No more template variables
            return result

        for var_name in matches:
            # Skip env.* variables (already handled)
            if var_name.startswith("env."):
                continue

            # Map context variables
            if var_name == "tmux_pane":
                value = context.get("pane")
            else:
                value = context.get(var_name)

            if value is None:
                # Required variable missing
                logger.debug("Template variable ${%s} not found in context", var_name)
                return ""

            # Replace ${var} with value
            result = result.replace(f"${{{var_name}}}", str(value))

        return result

register_rule(command, wrapper, owner)

Register a prefix rule programmatically (for Spokes/Gears).

Parameters:

Name	Type	Description	Default
command	str	Command to intercept (e.g., "ansible-playbook")	required
wrapper	str	Wrapper command (e.g., "axium ansible-run")	required
owner	str	Name of spoke/gear registering this rule	required

Returns:

Type	Description
bool	True if registered successfully, False if conflict detected

Example

>>> registry = PrefixRegistry()
>>> registry.register_rule("ansible-playbook", "axium ansible-run", "ansible")
True
>>> registry.register_rule("ansible-playbook", "other command", "terraform")
False  # Conflict!

Note

Logs warning and returns False on conflict. Use get_rule_owner() to check current owner before registering.

Source code in axium/core/prefix.py

def register_rule(self, command: str, wrapper: str, owner: str) -> bool:
    """
    Register a prefix rule programmatically (for Spokes/Gears).

    Args:
        command: Command to intercept (e.g., "ansible-playbook")
        wrapper: Wrapper command (e.g., "axium ansible-run")
        owner: Name of spoke/gear registering this rule

    Returns:
        True if registered successfully, False if conflict detected

    Example:
        ```python
        >>> registry = PrefixRegistry()
        >>> registry.register_rule("ansible-playbook", "axium ansible-run", "ansible")
        True
        >>> registry.register_rule("ansible-playbook", "other command", "terraform")
        False  # Conflict!
        ```

    Note:
        Logs warning and returns False on conflict.
        Use get_rule_owner() to check current owner before registering.
    """
    # Check for conflict
    if command in self.owners:
        existing_owner = self.owners[command]
        if existing_owner != owner:
            logger.warning(
                "Prefix conflict: '%s' tried to register command '%s' but already owned by '%s'",
                owner,
                command,
                existing_owner,
            )
            return False
        else:
            # Same owner re-registering (e.g., during reload) - allow and update
            logger.debug(
                "Updating prefix rule for command '%s' (owner: %s)", command, owner
            )
            # Remove old rule
            self.rules = [
                r
                for r in self.rules
                if r.get("command") != command or r.get("owner") != owner
            ]

    # Register new rule
    rule = {
        "command": command,
        "prefix": wrapper,  # Note: using 'prefix' key for consistency with yaml format
        "owner": owner,
    }
    self.rules.append(rule)
    self.owners[command] = owner
    logger.info(
        "Registered prefix rule: %s → %s (owner: %s)", command, wrapper, owner
    )
    return True

unregister_rules_for_owner(owner)

Unregister all prefix rules for a specific owner (spoke/gear).

Used when unloading a spoke/gear to clean up its registered rules.

Parameters:

Name	Type	Description	Default
owner	str	Name of spoke/gear to remove rules for	required

Returns:

Type	Description
int	Number of rules removed

Example

>>> registry.unregister_rules_for_owner("ansible")
2  # Removed 2 rules

Source code in axium/core/prefix.py

def unregister_rules_for_owner(self, owner: str) -> int:
    """
    Unregister all prefix rules for a specific owner (spoke/gear).

    Used when unloading a spoke/gear to clean up its registered rules.

    Args:
        owner: Name of spoke/gear to remove rules for

    Returns:
        Number of rules removed

    Example:
        ```python
        >>> registry.unregister_rules_for_owner("ansible")
        2  # Removed 2 rules
        ```
    """
    removed_commands = [
        cmd for cmd, cmd_owner in self.owners.items() if cmd_owner == owner
    ]
    self.rules = [r for r in self.rules if r.get("owner") != owner]

    for cmd in removed_commands:
        del self.owners[cmd]

    if removed_commands:
        logger.info(
            "Unregistered %d prefix rules for owner: %s",
            len(removed_commands),
            owner,
        )

    return len(removed_commands)

get_rule_owner(command)

Get the owner (spoke/gear name) of a prefix rule.

Parameters:

Name	Type	Description	Default
command	str	Command to check	required

Returns:

Type	Description
str | None	Owner name or None if no rule registered

Example

>>> registry.get_rule_owner("ansible-playbook")
'ansible'

Source code in axium/core/prefix.py

def get_rule_owner(self, command: str) -> str | None:
    """
    Get the owner (spoke/gear name) of a prefix rule.

    Args:
        command: Command to check

    Returns:
        Owner name or None if no rule registered

    Example:
        ```python
        >>> registry.get_rule_owner("ansible-playbook")
        'ansible'
        ```
    """
    return self.owners.get(command)

get_all_rules()

Get all registered prefix rules with metadata.

Returns:

Type	Description
list[dict[str, Any]]	List of rule dicts with command, prefix/wrapper, and owner

Example

>>> rules = registry.get_all_rules()
>>> for rule in rules:
...     print(f"{rule['command']} owned by {rule['owner']}")

Source code in axium/core/prefix.py

def get_all_rules(self) -> list[dict[str, Any]]:
    """
    Get all registered prefix rules with metadata.

    Returns:
        List of rule dicts with command, prefix/wrapper, and owner

    Example:
        ```python
        >>> rules = registry.get_all_rules()
        >>> for rule in rules:
        ...     print(f"{rule['command']} owned by {rule['owner']}")
        ```
    """
    return self.rules.copy()

get_prefixed_commands()

Get list of all commands that have prefix rules.

Source code in axium/core/prefix.py

def get_prefixed_commands(self) -> list[str]:
    """Get list of all commands that have prefix rules."""
    commands = set()
    for rule in self.rules:
        commands.add(rule["command"])
    return sorted(commands)

apply_prefixes(command, args, context)

Apply prefix rules to a command.

Parameters:

Name	Type	Description	Default
command	str	The command to prefix (e.g., "aws")	required
args	list[str]	Command arguments (e.g., ["s3", "ls"])	required
context	dict[str, Any]	Context dict with 'env', 'pane', etc.	required

Returns:

Name	Type	Description
	list[str]	Tuple of (command_array, env_vars)
command_array	dict[str, str]	Full command to execute (e.g., ["enva-root", "aws", "s3", "ls"])
env_vars	tuple[list[str], dict[str, str]]	Environment variables to set (currently unused, for future)

Source code in axium/core/prefix.py

def apply_prefixes(
    self, command: str, args: list[str], context: dict[str, Any]
) -> tuple[list[str], dict[str, str]]:
    """
    Apply prefix rules to a command.

    Args:
        command: The command to prefix (e.g., "aws")
        args: Command arguments (e.g., ["s3", "ls"])
        context: Context dict with 'env', 'pane', etc.

    Returns:
        Tuple of (command_array, env_vars)
        command_array: Full command to execute
            (e.g., ["enva-root", "aws", "s3", "ls"])
        env_vars: Environment variables to set
            (currently unused, for future)
    """
    # Find matching rules for this command
    matching_rules = [r for r in self.rules if r["command"] == command]

    if not matching_rules:
        # No rules, return unwrapped command
        return ([command] + args, {})

    # Apply first matching rule (TODO: support multiple rules/conditions)
    rule = matching_rules[0]
    prefix_template = rule["prefix"]

    # Expand template variables
    expanded_prefix = self._expand_template(prefix_template, context)

    if not expanded_prefix:
        # Template expansion failed (e.g., missing variable)
        logger.warning("Failed to expand prefix template: %s", prefix_template)
        return ([command] + args, {})

    # Return prefixed command
    final_command = [expanded_prefix, command] + args
    logger.debug("Prefixed command: %s", " ".join(final_command))
    return (final_command, {})

expand_env_vars(template)

Expand ${env.} template variables using active environment data.

This function resolves environment property references in templates by looking them up in the active environment from envs.yaml.

Parameters:

Name	Type	Description	Default
template	str	Template string with ${env.} patterns (e.g., "Using ${env.prefix} in ${env.region}")	required

Returns:

Type	Description
str	Expanded string with variables replaced, or original string if
str	no active environment or variables not found.

Example

>>> # Assuming active env is "root" with prefix="enva-root"
>>> expand_env_vars("${env.prefix}")
'enva-root'
>>> expand_env_vars("Using ${env.prefix} in ${env.region}")
'Using enva-root in eu-west-1'
>>> expand_env_vars("No variables here")
'No variables here'

Note

This function now wraps axium.core.utils.expand_template for consistency.

Source code in axium/core/prefix.py

def expand_env_vars(template: str) -> str:
    """
    Expand ${env.<key>} template variables using active environment data.

    This function resolves environment property references in templates by
    looking them up in the active environment from envs.yaml.

    Args:
        template: Template string with ${env.<key>} patterns
                 (e.g., "Using ${env.prefix} in ${env.region}")

    Returns:
        Expanded string with variables replaced, or original string if
        no active environment or variables not found.

    Example:
        ```python
        >>> # Assuming active env is "root" with prefix="enva-root"
        >>> expand_env_vars("${env.prefix}")
        'enva-root'
        >>> expand_env_vars("Using ${env.prefix} in ${env.region}")
        'Using enva-root in eu-west-1'
        >>> expand_env_vars("No variables here")
        'No variables here'
        ```

    Note:
        This function now wraps axium.core.utils.expand_template for consistency.
    """
    from axium.core.utils import expand_template

    return expand_template(
        template,
        env_dict=None,  # Will auto-load from active environment
        expand_tilde=False,  # Only expand env vars
        expand_env_vars=False,  # Don't expand $VAR
        expand_env_keys=True,  # Only expand ${env.key}
    )

get_registry()

Get the global prefix registry (singleton).

Source code in axium/core/prefix.py

def get_registry() -> PrefixRegistry:
    """Get the global prefix registry (singleton)."""
    global _registry
    if _registry is None:
        _registry = PrefixRegistry()
    return _registry

apply_prefixes(command, args, context)

Apply prefix rules to a command.

Source code in axium/core/prefix.py

def apply_prefixes(
    command: str, args: list[str], context: dict[str, Any]
) -> tuple[list[str], dict[str, str]]:
    """Apply prefix rules to a command."""
    return get_registry().apply_prefixes(command, args, context)

get_prefixed_commands()

Get list of all commands with prefix rules.

Source code in axium/core/prefix.py

def get_prefixed_commands() -> list[str]:
    """Get list of all commands with prefix rules."""
    return get_registry().get_prefixed_commands()

register_prefix_rule(command, wrapper, owner)

Register a prefix rule (for Spokes/Gears to call).

Parameters:

Name	Type	Description	Default
command	str	Command to intercept	required
wrapper	str	Wrapper command to execute instead	required
owner	str	Name of spoke/gear registering the rule	required

Returns:

Type	Description
bool	True if registered, False if conflict detected

Example

>>> from axium.core import prefix
>>> prefix.register_prefix_rule("terraform", "axium tf-run", "terraform-gear")
True

Source code in axium/core/prefix.py

def register_prefix_rule(command: str, wrapper: str, owner: str) -> bool:
    """
    Register a prefix rule (for Spokes/Gears to call).

    Args:
        command: Command to intercept
        wrapper: Wrapper command to execute instead
        owner: Name of spoke/gear registering the rule

    Returns:
        True if registered, False if conflict detected

    Example:
        ```python
        >>> from axium.core import prefix
        >>> prefix.register_prefix_rule("terraform", "axium tf-run", "terraform-gear")
        True
        ```
    """
    return get_registry().register_rule(command, wrapper, owner)

unregister_rules_for_owner(owner)

Unregister all prefix rules for an owner.

Parameters:

Name	Type	Description	Default
owner	str	Name of spoke/gear	required

Returns:

Type	Description
int	Number of rules removed

Source code in axium/core/prefix.py

def unregister_rules_for_owner(owner: str) -> int:
    """
    Unregister all prefix rules for an owner.

    Args:
        owner: Name of spoke/gear

    Returns:
        Number of rules removed
    """
    return get_registry().unregister_rules_for_owner(owner)

get_rule_owner(command)

Get the owner of a prefix rule.

Source code in axium/core/prefix.py

def get_rule_owner(command: str) -> str | None:
    """Get the owner of a prefix rule."""
    return get_registry().get_rule_owner(command)

get_all_rules()

Get all registered prefix rules.

Source code in axium/core/prefix.py

def get_all_rules() -> list[dict[str, Any]]:
    """Get all registered prefix rules."""
    return get_registry().get_all_rules()

reload_config()

Reload prefixes.yaml configuration.

Source code in axium/core/prefix.py

def reload_config() -> None:
    """Reload prefixes.yaml configuration."""
    global _registry
    _registry = PrefixRegistry()
    logger.info("Reloaded prefix configuration")

---

bootstrap

Self-bootstrapping initialization system.

bootstrap

Bootstrap module for Axium self-initialization.

Ensures Axium config directory and default files exist on first run. Handles graceful initialization without overwriting existing user data.

update_init_scripts()

Force update shell integration scripts (bash/init.sh, tmux/init.sh).

This function overwrites existing init scripts with the latest versions from the package. It does NOT modify user configuration files.

Use this after upgrading Axium to get the latest shell integrations.

Raises:

Type	Description
Exception	If script update fails

Source code in axium/core/bootstrap.py

def update_init_scripts() -> None:
    """
    Force update shell integration scripts (bash/init.sh, tmux/init.sh).

    This function overwrites existing init scripts with the latest versions
    from the package. It does NOT modify user configuration files.

    Use this after upgrading Axium to get the latest shell integrations.

    Raises:
        Exception: If script update fails
    """
    try:
        # Ensure directories exist
        BASH_DIR.mkdir(parents=True, exist_ok=True)
        TMUX_DIR.mkdir(parents=True, exist_ok=True)

        # Remove old scripts if they exist
        if BASH_INIT_PATH.exists():
            BASH_INIT_PATH.unlink()
        if TMUX_INIT_PATH.exists():
            TMUX_INIT_PATH.unlink()

        # Create fresh scripts
        _create_bash_init()
        _create_tmux_init()

        logger.info("Updated shell integration scripts")
    except Exception as e:
        logger.error("Failed to update init scripts: %s", e)
        raise

ensure_axium_config()

Ensure Axium config directory and files exist.

Creates ~/.config/axium/ with default configuration files if missing. Never overwrites existing files.

Creates

• ~/.config/axium/ (directory)
• envs.yaml (with root/builder defaults)
• prefixes.yaml (with aws/terraform defaults)
• hud.yaml (HUD layout and style configuration)
• state.json (with active_env: root)
• state_cache.json (cached prefixed commands for fast shell startup)
• spokes/ (empty directory)
• bash/init.sh (bash/zsh shell integration script)
• tmux/init.sh (tmux integration script)

Returns:

Type	Description
bool	True if initialization was performed (first run)
bool	False if config already existed

Raises:

Type	Description
SystemExit	If config directory cannot be created (permission denied)

Source code in axium/core/bootstrap.py

def ensure_axium_config() -> bool:
    """
    Ensure Axium config directory and files exist.

    Creates ~/.config/axium/ with default configuration files if missing.
    Never overwrites existing files.

    Creates:
        - ~/.config/axium/ (directory)
        - envs.yaml (with root/builder defaults)
        - prefixes.yaml (with aws/terraform defaults)
        - hud.yaml (HUD layout and style configuration)
        - state.json (with active_env: root)
        - state_cache.json (cached prefixed commands for fast shell startup)
        - spokes/ (empty directory)
        - bash/init.sh (bash/zsh shell integration script)
        - tmux/init.sh (tmux integration script)

    Returns:
        True if initialization was performed (first run)
        False if config already existed

    Raises:
        SystemExit: If config directory cannot be created (permission denied)
    """
    # Check if config directory exists and create if needed
    try:
        config_existed = CONF_DIR.exists()
        CONF_DIR.mkdir(parents=True, exist_ok=True)
    except PermissionError:
        logger.error(
            "axium: cannot create config directory at %s (permission denied)",
            CONF_DIR,
        )
        sys.exit(1)
    except Exception as e:
        logger.error("axium: config directory creation failed: %s", e)
        sys.exit(1)

    # Track if any initialization was needed
    initialized = False

    # Create default files if they don't exist
    try:
        if not ENVS_PATH.exists():
            _create_default_envs()
            initialized = True

        if not PREFIXES_PATH.exists():
            _create_default_prefixes()
            initialized = True

        if not STATE_PATH.exists():
            _create_default_state()
            initialized = True

        if not HUD_PATH.exists():
            _create_default_hud()
            initialized = True

        if not SPOKES_DIR.exists():
            _ensure_spokes_dir()
            initialized = True

        if not BASH_INIT_PATH.exists():
            _create_bash_init()
            initialized = True

        if not TMUX_INIT_PATH.exists():
            _create_tmux_init()
            initialized = True

    except Exception as e:
        logger.error("axium: config initialization failed: %s", e)
        sys.exit(1)

    # Log message only if initialization was performed
    if initialized or not config_existed:
        logger.info("axium: initialized config at %s", CONF_DIR)
        return True

    return False

---

ipc

Inter-process communication utilities for daemon/CLI communication.

ipc

Axium IPC - Inter-process communication utilities.

Provides functions for CLI to communicate with the daemon via UNIX socket. Messages are JSON-encoded and sent over /tmp/axiumd.sock.

Protocol

• Line-based JSON protocol (newline-delimited)
• Request format:

  {"cmd": "command_name", "arg": "value", ...}
  

• Response format:

  {"ok": true, ...}
  // or
  {"ok": false, "error": "..."}
  

• Connection closed after each request/response

Example

>>> send_request_sync({"cmd": "ping"})
{'ok': True, 'pong': True}

>>> send_request_sync({"cmd": "get_state"})
{'ok': True, 'state': {'active_env': 'prod', ...}}

>>> send_request_sync({"cmd": "set_env", "value": "prod"})
{'ok': True}

send_request_sync(payload, timeout=2.0)

Send IPC request to daemon (synchronous wrapper).

This is the primary IPC function used by CLI commands. Wraps the async _send() implementation with asyncio.run() for synchronous use.

Parameters:

Name	Type	Description	Default
payload	dict	Dict to send as JSON (e.g., {"cmd": "get_state"})	required
timeout	float	Maximum seconds to wait for response (default: 2.0)	2.0

Returns:

Type	Description
dict	Response dict from daemon

Raises:

Type	Description
FileNotFoundError	If daemon socket doesn't exist
ConnectionRefusedError	If daemon not accepting connections
JSONDecodeError	If daemon response is malformed
TimeoutError	If daemon doesn't respond within timeout
Exception	Any other communication errors

Example

>>> send_request_sync({"cmd": "ping"})
{'ok': True, 'pong': True}

>>> send_request_sync({"cmd": "set_env", "value": "prod"})
{'ok': True}

>>> send_request_sync({"cmd": "get_state"})
{'ok': True, 'state': {'active_env': 'prod', 'started': '...'}}

>>> send_request_sync({"cmd": "slow_op"}, timeout=10.0)
{'ok': True}

Common Commands

ping: Health check send_request_sync({"cmd": "ping"})

get_state: Get daemon state send_request_sync({"cmd": "get_state"})

set_env: Set active environment send_request_sync({"cmd": "set_env", "value": "prod"})

reload: Reload state from disk send_request_sync({"cmd": "reload"})

apply_prefixes: Apply prefix rules send_request_sync({ "cmd": "apply_prefixes", "command": "aws", "args": ["s3", "ls"], "context": {"pane": "...", "env": "prod"} })

stop: Gracefully stop daemon send_request_sync({"cmd": "stop"})

Note

Default timeout is 2 seconds for quick failure when daemon is down. Each call opens and closes a new connection (no connection pooling).

Source code in axium/core/ipc.py

def send_request_sync(payload: dict, timeout: float = 2.0) -> dict:
    """
    Send IPC request to daemon (synchronous wrapper).

    This is the primary IPC function used by CLI commands. Wraps the
    async _send() implementation with asyncio.run() for synchronous use.

    Args:
        payload: Dict to send as JSON (e.g., {"cmd": "get_state"})
        timeout: Maximum seconds to wait for response (default: 2.0)

    Returns:
        Response dict from daemon

    Raises:
        FileNotFoundError: If daemon socket doesn't exist
        ConnectionRefusedError: If daemon not accepting connections
        json.JSONDecodeError: If daemon response is malformed
        asyncio.TimeoutError: If daemon doesn't respond within timeout
        Exception: Any other communication errors

    Example:
        ```python
        >>> send_request_sync({"cmd": "ping"})
        {'ok': True, 'pong': True}

        >>> send_request_sync({"cmd": "set_env", "value": "prod"})
        {'ok': True}

        >>> send_request_sync({"cmd": "get_state"})
        {'ok': True, 'state': {'active_env': 'prod', 'started': '...'}}

        >>> send_request_sync({"cmd": "slow_op"}, timeout=10.0)
        {'ok': True}
        ```

    Common Commands:
        ping: Health check
            send_request_sync({"cmd": "ping"})

        get_state: Get daemon state
            send_request_sync({"cmd": "get_state"})

        set_env: Set active environment
            send_request_sync({"cmd": "set_env", "value": "prod"})

        reload: Reload state from disk
            send_request_sync({"cmd": "reload"})

        apply_prefixes: Apply prefix rules
            send_request_sync({
                "cmd": "apply_prefixes",
                "command": "aws",
                "args": ["s3", "ls"],
                "context": {"pane": "...", "env": "prod"}
            })

        stop: Gracefully stop daemon
            send_request_sync({"cmd": "stop"})

    Note:
        Default timeout is 2 seconds for quick failure when daemon is down.
        Each call opens and closes a new connection (no connection pooling).
    """
    return asyncio.run(asyncio.wait_for(_send(payload), timeout=timeout))