Spokes API

Plugin system and event bus.

---

spokes

Spoke discovery, loading, and event bus implementation.

spokes

Axium Spokes - Plugin system and event bus.

Spokes are the plugin architecture for Axium, allowing modular extensions without modifying core code. Each Spoke lives in ~/.config/axium/spokes// with a spoke.yaml manifest and Python module.

The EventBus provides event-driven coordination, allowing Spokes to: - React to environment changes (env_change event) - Know when other Spokes load (spoke_loaded event) - Add custom CLI commands via Typer - Register prefix rules dynamically

Spoke Structure

~/.config/axium/spokes/aws/ spoke.yaml - Manifest with name and entrypoint main.py - Implementation with register(app, events) function

Manifest Format (spoke.yaml): name: aws entrypoint: aws.main:register

Standard Events

• env_change(new_env: str, old_env: str) - Environment switched
• spoke_loaded(spoke_name: str) - Spoke finished loading

Example Spoke

def register(app, events): @app.command("aws-whoami") def aws_whoami(): import boto3 print(boto3.client("sts").get_caller_identity())

def on_env_change(new_env, old_env):
    print(f"Env changed: {old_env} → {new_env}")

events.on("env_change", on_env_change)

SpokeMetadata dataclass

Metadata for an installed Spoke.

Tracks installation details, version, source, and current status for lifecycle management and display purposes.

Attributes:

Name	Type	Description
name	str	Spoke name (matches directory name)
version	str	Semantic version string (e.g., "0.1.0")
description	str	Short description of Spoke functionality
entrypoint	str	Module:function for register() (e.g., "aws.main:register")
source	str	Installation source - "local:/path", "git:url", "registry:name"
install_mode	str	"copy" or "symlink" (symlink for editable installs)
installed_at	str	ISO 8601 timestamp of installation
last_loaded	str | None	ISO 8601 timestamp of last successful load (None if never loaded)
status	str	Current status - "active", "error", or "not-loaded"

Example

>>> metadata = SpokeMetadata(
...     name="aws",
...     version="0.1.0",
...     description="AWS helpers",
...     entrypoint="aws.main:register",
...     source="local:/Users/jon/spoke-aws",
...     install_mode="symlink",
...     installed_at="2025-10-07T15:30:00Z",
...     last_loaded="2025-10-07T15:31:00Z",
...     status="active"
... )

Source code in axium/core/spokes.py

@dataclass
class SpokeMetadata:
    """
    Metadata for an installed Spoke.

    Tracks installation details, version, source, and current status
    for lifecycle management and display purposes.

    Attributes:
        name: Spoke name (matches directory name)
        version: Semantic version string (e.g., "0.1.0")
        description: Short description of Spoke functionality
        entrypoint: Module:function for register() (e.g., "aws.main:register")
        source: Installation source - "local:/path", "git:url", "registry:name"
        install_mode: "copy" or "symlink" (symlink for editable installs)
        installed_at: ISO 8601 timestamp of installation
        last_loaded: ISO 8601 timestamp of last successful load (None if never loaded)
        status: Current status - "active", "error", or "not-loaded"

    Example:
        ```python
        >>> metadata = SpokeMetadata(
        ...     name="aws",
        ...     version="0.1.0",
        ...     description="AWS helpers",
        ...     entrypoint="aws.main:register",
        ...     source="local:/Users/jon/spoke-aws",
        ...     install_mode="symlink",
        ...     installed_at="2025-10-07T15:30:00Z",
        ...     last_loaded="2025-10-07T15:31:00Z",
        ...     status="active"
        ... )
        ```
    """

    name: str
    version: str
    description: str
    entrypoint: str
    source: str
    install_mode: str
    installed_at: str
    last_loaded: str | None = None
    status: str = "not-loaded"

    def to_dict(self) -> dict[str, Any]:
        """
        Convert metadata to dictionary for JSON serialization.

        Returns:
            Dict with all metadata fields

        Example:
            ```python
            >>> metadata.to_dict()
            {'name': 'aws', 'version': '0.1.0', ...}
            ```
        """
        return asdict(self)

to_dict()

Convert metadata to dictionary for JSON serialization.

Returns:

Type	Description
dict[str, Any]	Dict with all metadata fields

Example

>>> metadata.to_dict()
{'name': 'aws', 'version': '0.1.0', ...}

Source code in axium/core/spokes.py

def to_dict(self) -> dict[str, Any]:
    """
    Convert metadata to dictionary for JSON serialization.

    Returns:
        Dict with all metadata fields

    Example:
        ```python
        >>> metadata.to_dict()
        {'name': 'aws', 'version': '0.1.0', ...}
        ```
    """
    return asdict(self)

EventBus

Event bus for Spoke coordination.

Provides publish-subscribe pattern for loosely-coupled communication between Axium core and Spokes. Events are synchronous and handlers are called in registration order.

Standard Events

env_change(new_env, old_env, pane=None): Emitted when environment switches spoke_loaded(spoke_name): Emitted when a Spoke finishes initial load spoke_reloaded(spoke_name): Emitted after a Spoke is reloaded (post-action) spoke_unloaded(spoke_name): Emitted after a Spoke is unloaded (post-action) gear_loaded(gear_name): Emitted when a Gear finishes loading gear_unloaded(gear_name): Emitted after a Gear is unloaded daemon_reload: Emitted after daemon configuration is reloaded config_reloaded: Emitted after all spoke configs are cleared and reloaded hud_segment_updated(spoke, value): Emitted when a HUD segment changes hud_refresh: Emitted when HUD should be regenerated

Attributes:

Name	Type	Description
_listeners	dict[str, list[Callable]]	Dict mapping event names to lists of callback functions

Example

>>> bus = EventBus()
>>> def handler(new, old):
...     print(f"Changed: {old} → {new}")
>>> bus.on("env_change", handler)
>>> bus.emit("env_change", "prod", "dev")
Changed: dev → prod

Source code in axium/core/spokes.py

class EventBus:
    """
    Event bus for Spoke coordination.

    Provides publish-subscribe pattern for loosely-coupled communication
    between Axium core and Spokes. Events are synchronous and handlers
    are called in registration order.

    Standard Events:
        env_change(new_env, old_env, pane=None): Emitted when environment switches
        spoke_loaded(spoke_name): Emitted when a Spoke finishes initial load
        spoke_reloaded(spoke_name): Emitted after a Spoke is reloaded (post-action)
        spoke_unloaded(spoke_name): Emitted after a Spoke is unloaded (post-action)
        gear_loaded(gear_name): Emitted when a Gear finishes loading
        gear_unloaded(gear_name): Emitted after a Gear is unloaded
        daemon_reload: Emitted after daemon configuration is reloaded
        config_reloaded: Emitted after all spoke configs are cleared and reloaded
        hud_segment_updated(spoke, value): Emitted when a HUD segment changes
        hud_refresh: Emitted when HUD should be regenerated

    Attributes:
        _listeners: Dict mapping event names to lists of callback functions

    Example:
        ```python
        >>> bus = EventBus()
        >>> def handler(new, old):
        ...     print(f"Changed: {old} → {new}")
        >>> bus.on("env_change", handler)
        >>> bus.emit("env_change", "prod", "dev")
        Changed: dev → prod
        ```
    """

    def __init__(self):
        """
        Initialize empty event bus.

        Creates an empty listener registry. Listeners are added via on().
        """
        self._listeners: dict[str, list[Callable]] = {}

    def on(self, event_name: str, callback: Callable) -> None:
        """
        Subscribe to an event.

        Registers a callback to be invoked when the event is emitted.
        Multiple callbacks can be registered for the same event.

        Args:
            event_name: Name of event to listen for (e.g., "env_change")
            callback: Function to call when event is emitted

        Example:
            ```python
            >>> def on_env_change(new_env, old_env):
            ...     print(f"Environment: {new_env}")
            >>> events.on("env_change", on_env_change)
            ```

        Note:
            Callbacks are invoked synchronously in registration order.
            Exceptions in callbacks are logged but don't stop event propagation.
        """
        if event_name not in self._listeners:
            self._listeners[event_name] = []
        self._listeners[event_name].append(callback)
        logger.debug("Registered listener for event: %s", event_name)

    def emit(self, event_name: str, *args, **kwargs) -> None:
        """
        Emit an event to all subscribers.

        Calls all registered callbacks for this event with provided arguments.
        If a callback raises an exception, it's logged and remaining callbacks
        continue to execute.

        Args:
            event_name: Name of event to emit
            *args: Positional arguments to pass to callbacks
            **kwargs: Keyword arguments to pass to callbacks

        Example:
            ```python
            >>> events.emit("env_change", "prod", "dev")
            >>> events.emit("custom_event", key="value")
            ```

        Note:
            Events are synchronous - emit() blocks until all callbacks complete.
            Unknown events (no listeners) are silently ignored.
        """
        listeners = self._listeners.get(event_name, [])
        logger.debug("Emitting event: %s to %d listeners", event_name, len(listeners))
        for callback in listeners:
            try:
                callback(*args, **kwargs)
            except Exception as e:
                logger.error("Event listener failed for %s: %s", event_name, e)

__init__()

Initialize empty event bus.

Creates an empty listener registry. Listeners are added via on().

Source code in axium/core/spokes.py

def __init__(self):
    """
    Initialize empty event bus.

    Creates an empty listener registry. Listeners are added via on().
    """
    self._listeners: dict[str, list[Callable]] = {}

on(event_name, callback)

Subscribe to an event.

Registers a callback to be invoked when the event is emitted. Multiple callbacks can be registered for the same event.

Parameters:

Name	Type	Description	Default
event_name	str	Name of event to listen for (e.g., "env_change")	required
callback	Callable	Function to call when event is emitted	required

Example

>>> def on_env_change(new_env, old_env):
...     print(f"Environment: {new_env}")
>>> events.on("env_change", on_env_change)

Note

Callbacks are invoked synchronously in registration order. Exceptions in callbacks are logged but don't stop event propagation.

Source code in axium/core/spokes.py

def on(self, event_name: str, callback: Callable) -> None:
    """
    Subscribe to an event.

    Registers a callback to be invoked when the event is emitted.
    Multiple callbacks can be registered for the same event.

    Args:
        event_name: Name of event to listen for (e.g., "env_change")
        callback: Function to call when event is emitted

    Example:
        ```python
        >>> def on_env_change(new_env, old_env):
        ...     print(f"Environment: {new_env}")
        >>> events.on("env_change", on_env_change)
        ```

    Note:
        Callbacks are invoked synchronously in registration order.
        Exceptions in callbacks are logged but don't stop event propagation.
    """
    if event_name not in self._listeners:
        self._listeners[event_name] = []
    self._listeners[event_name].append(callback)
    logger.debug("Registered listener for event: %s", event_name)

emit(event_name, *args, **kwargs)

Emit an event to all subscribers.

Calls all registered callbacks for this event with provided arguments. If a callback raises an exception, it's logged and remaining callbacks continue to execute.

Parameters:

Name	Type	Description	Default
event_name	str	Name of event to emit	required
*args		Positional arguments to pass to callbacks	()
**kwargs		Keyword arguments to pass to callbacks	{}

Example

>>> events.emit("env_change", "prod", "dev")
>>> events.emit("custom_event", key="value")

Note

Events are synchronous - emit() blocks until all callbacks complete. Unknown events (no listeners) are silently ignored.

Source code in axium/core/spokes.py

def emit(self, event_name: str, *args, **kwargs) -> None:
    """
    Emit an event to all subscribers.

    Calls all registered callbacks for this event with provided arguments.
    If a callback raises an exception, it's logged and remaining callbacks
    continue to execute.

    Args:
        event_name: Name of event to emit
        *args: Positional arguments to pass to callbacks
        **kwargs: Keyword arguments to pass to callbacks

    Example:
        ```python
        >>> events.emit("env_change", "prod", "dev")
        >>> events.emit("custom_event", key="value")
        ```

    Note:
        Events are synchronous - emit() blocks until all callbacks complete.
        Unknown events (no listeners) are silently ignored.
    """
    listeners = self._listeners.get(event_name, [])
    logger.debug("Emitting event: %s to %d listeners", event_name, len(listeners))
    for callback in listeners:
        try:
            callback(*args, **kwargs)
        except Exception as e:
            logger.error("Event listener failed for %s: %s", event_name, e)

get_event_bus()

Get the global event bus singleton.

Returns the shared EventBus instance used by core and all Spokes. This ensures all parts of Axium communicate via the same event bus.

Returns:

Type	Description
EventBus	Global EventBus instance

Example

>>> from axium.core import spokes
>>> events = spokes.get_event_bus()
>>> events.on("env_change", my_handler)

Note

The singleton is created on first module import. All Spokes receive the same instance via register(app, events).

Source code in axium/core/spokes.py

def get_event_bus() -> EventBus:
    """
    Get the global event bus singleton.

    Returns the shared EventBus instance used by core and all Spokes.
    This ensures all parts of Axium communicate via the same event bus.

    Returns:
        Global EventBus instance

    Example:
        ```python
        >>> from axium.core import spokes
        >>> events = spokes.get_event_bus()
        >>> events.on("env_change", my_handler)
        ```

    Note:
        The singleton is created on first module import.
        All Spokes receive the same instance via register(app, events).
    """
    return _event_bus

get_spoke_metadata()

Get metadata for all installed Spokes.

Loads metadata from JSON file and enriches with current status from the command registry.

Returns:

Type	Description
list[SpokeMetadata]	List of SpokeMetadata objects, sorted by name

Example

>>> spokes = get_spoke_metadata()
>>> for spoke in spokes:
...     print(f"{spoke.name}: {spoke.status}")
aws: active
k8s: active

Note

Status is enriched with current load state from registry. Spokes in metadata but not in registry show as "not-loaded".

Source code in axium/core/spokes.py

def get_spoke_metadata() -> list[SpokeMetadata]:
    """
    Get metadata for all installed Spokes.

    Loads metadata from JSON file and enriches with current status
    from the command registry.

    Returns:
        List of SpokeMetadata objects, sorted by name

    Example:
        ```python
        >>> spokes = get_spoke_metadata()
        >>> for spoke in spokes:
        ...     print(f"{spoke.name}: {spoke.status}")
        aws: active
        k8s: active
        ```

    Note:
        Status is enriched with current load state from registry.
        Spokes in metadata but not in registry show as "not-loaded".
    """
    data = _load_metadata_file()
    spokes_data = data.get("spokes", {})

    # Convert dict entries to SpokeMetadata objects
    metadata_list = []
    for spoke_name, spoke_dict in spokes_data.items():
        try:
            metadata = SpokeMetadata(**spoke_dict)
            metadata_list.append(metadata)
        except Exception as e:
            logger.error("Invalid metadata for Spoke %s: %s", spoke_name, e)

    # Sort by name
    metadata_list.sort(key=lambda s: s.name)

    return metadata_list

create_spoke(name, description=None, version='0.1.0', interactive=True)

Create a new Spoke from template.

Scaffolds a new Spoke directory with spoke.yaml manifest and main.py implementation template. Prompts for details interactively if enabled.

Parameters:

Name	Type	Description	Default
name	str	Spoke name (used for directory and defaults)	required
description	str | None	Short description (prompted if None and interactive=True)	None
version	str	Semantic version string (default: "0.1.0")	'0.1.0'
interactive	bool	Prompt for missing values if True	True

Returns:

Type	Description
Path	Path to created Spoke directory

Example

$ axium spoke new aws
Creating Spoke: aws
Description: AWS helpers for Axium
Version [0.1.0]:
✓ Directory: ~/.config/axium/spokes/aws/
✓ Generated spoke.yaml
✓ Generated main.py

Raises:

Type	Description
FileExistsError	If Spoke directory already exists
ValueError	If name is invalid (empty, contains invalid characters)

Note

• Creates directory structure in ~/.config/axium/spokes//
• Generates spoke.yaml with metadata
• Creates main.py with register() function and example command
• Adds entry to .spoke_metadata.json

Source code in axium/core/spokes.py

def create_spoke(
    name: str,
    description: str | None = None,
    version: str = "0.1.0",
    interactive: bool = True,
) -> Path:
    """
    Create a new Spoke from template.

    Scaffolds a new Spoke directory with spoke.yaml manifest and
    main.py implementation template. Prompts for details interactively
    if enabled.

    Args:
        name: Spoke name (used for directory and defaults)
        description: Short description (prompted if None and interactive=True)
        version: Semantic version string (default: "0.1.0")
        interactive: Prompt for missing values if True

    Returns:
        Path to created Spoke directory

    Example:
        ```bash
        $ axium spoke new aws
        Creating Spoke: aws
        Description: AWS helpers for Axium
        Version [0.1.0]:
        ✓ Directory: ~/.config/axium/spokes/aws/
        ✓ Generated spoke.yaml
        ✓ Generated main.py
        ```

    Raises:
        FileExistsError: If Spoke directory already exists
        ValueError: If name is invalid (empty, contains invalid characters)

    Note:
        - Creates directory structure in ~/.config/axium/spokes/<name>/
        - Generates spoke.yaml with metadata
        - Creates main.py with register() function and example command
        - Adds entry to .spoke_metadata.json
    """
    # Validate name
    if not name or not name.replace("-", "").replace("_", "").isalnum():
        raise ValueError(f"Invalid Spoke name: {name}")

    spoke_dir = SPOKES_DIR / name

    # Check if already exists
    if spoke_dir.exists():
        raise FileExistsError(f"Spoke directory already exists: {spoke_dir}")

    # Interactive prompts
    if interactive:
        import typer

        if description is None:
            description = typer.prompt(
                f"Description for '{name}'", default=f"{name} Spoke for Axium"
            )

        version_input = typer.prompt("Version", default=version)
        if version_input:
            version = version_input

    # Use defaults if still None
    if description is None:
        description = f"{name} Spoke for Axium"

    # Get author from git config or use placeholder
    author = "Your Name"
    try:
        import subprocess

        result = subprocess.run(
            ["git", "config", "user.name"], capture_output=True, text=True, timeout=2
        )
        if result.returncode == 0:
            author = result.stdout.strip()
    except Exception:
        pass

    # Create directory
    spoke_dir.mkdir(parents=True, exist_ok=True)
    logger.info("Created Spoke directory: %s", spoke_dir)

    # Write spoke.yaml
    spoke_yaml = spoke_dir / "spoke.yaml"
    spoke_yaml.write_text(
        SPOKE_YAML_TEMPLATE.format(
            name=name, version=version, description=description, author=author
        )
    )
    logger.debug("Generated spoke.yaml")

    # Write main.py
    main_py = spoke_dir / "main.py"
    main_py.write_text(
        SPOKE_MAIN_PY_TEMPLATE.format(name=name, description=description)
    )
    logger.debug("Generated main.py")

    # Add to metadata
    metadata_data = _load_metadata_file()
    metadata_data["spokes"][name] = SpokeMetadata(
        name=name,
        version=version,
        description=description,
        entrypoint=f"{name}.main:register",
        source=f"local:{spoke_dir}",
        install_mode="created",
        installed_at=datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
        status="not-loaded",
    ).to_dict()
    _save_metadata_file(metadata_data)

    print(f"✓ Spoke '{name}' created at {spoke_dir}")
    print(f"\nNext steps:")
    print(f"  1. Edit {spoke_dir}/main.py")
    print(f"  2. Run: axium spoke reload")
    print(f"  3. Test: axium {name}-example")

    return spoke_dir

validate_spoke(spoke_dir)

Validate Spoke directory structure and manifest.

Checks that spoke_dir contains a valid spoke.yaml with required fields and that the entrypoint module exists.

Parameters:

Name	Type	Description	Default
spoke_dir	Path	Path to Spoke directory to validate	required

Returns:

Type	Description
dict[str, Any]	Dict with parsed spoke.yaml data if valid

Example

>>> data = validate_spoke(Path("~/.config/axium/spokes/aws"))
>>> data["name"]
'aws'

Raises:

Type	Description
FileNotFoundError	If spoke.yaml missing
ValueError	If spoke.yaml invalid or missing required fields

Note

Required fields in spoke.yaml: name, version, description, entrypoint Entrypoint format: "spoke.module:function" (e.g., "aws.main:register")

Source code in axium/core/spokes.py

def validate_spoke(spoke_dir: Path) -> dict[str, Any]:
    """
    Validate Spoke directory structure and manifest.

    Checks that spoke_dir contains a valid spoke.yaml with required fields
    and that the entrypoint module exists.

    Args:
        spoke_dir: Path to Spoke directory to validate

    Returns:
        Dict with parsed spoke.yaml data if valid

    Example:
        ```python
        >>> data = validate_spoke(Path("~/.config/axium/spokes/aws"))
        >>> data["name"]
        'aws'
        ```

    Raises:
        FileNotFoundError: If spoke.yaml missing
        ValueError: If spoke.yaml invalid or missing required fields

    Note:
        Required fields in spoke.yaml: name, version, description, entrypoint
        Entrypoint format: "spoke.module:function" (e.g., "aws.main:register")
    """
    manifest_path = spoke_dir / "spoke.yaml"

    if not manifest_path.exists():
        raise FileNotFoundError(f"spoke.yaml not found in {spoke_dir}")

    try:
        data = yaml.safe_load(manifest_path.read_text())
    except Exception as e:
        raise ValueError(f"Invalid YAML in spoke.yaml: {e}")

    # Check required fields
    required_fields = ["name", "version", "description", "entrypoint"]
    missing = [f for f in required_fields if f not in data]
    if missing:
        raise ValueError(f"spoke.yaml missing required fields: {', '.join(missing)}")

    # Validate entrypoint format
    entrypoint = data.get("entrypoint", "")
    if ":" not in entrypoint:
        raise ValueError(
            f"Invalid entrypoint format '{entrypoint}' - must be 'module:function'"
        )

    module_part, func_part = entrypoint.split(":", 1)
    if not module_part or not func_part:
        raise ValueError(f"Invalid entrypoint format '{entrypoint}'")

    # Check if entrypoint file exists (if it's a .py file reference)
    if module_part.endswith(".py"):
        entrypoint_file = spoke_dir / module_part
        if not entrypoint_file.exists():
            raise FileNotFoundError(f"Entrypoint file not found: {entrypoint_file}")

    return data

install_spoke(source, editable=False, name=None)

Install a Spoke from source.

Supports installation from local directories. Git and registry sources are planned for future implementation.

Parameters:

Name	Type	Description	Default
source	str	Installation source - local path, git URL, or registry name	required
editable	bool	If True, symlink instead of copy (for development)	False
name	str | None	Override Spoke name from manifest (optional)	None

Returns:

Type	Description
SpokeMetadata	SpokeMetadata object for installed Spoke

Example

# Install from local directory
$ axium spoke install ~/my-spokes/aws

# Install in editable mode (symlink)
$ axium spoke install ~/dev/my-spoke --editable

Raises:

Type	Description
FileNotFoundError	If source doesn't exist or invalid
ValueError	If spoke.yaml invalid
FileExistsError	If Spoke already installed

Note

• Validates spoke.yaml before installation
• Adds entry to .spoke_metadata.json
• Does not auto-reload (caller should call reload_spokes())

Source code in axium/core/spokes.py

def install_spoke(
    source: str, editable: bool = False, name: str | None = None
) -> SpokeMetadata:
    """
    Install a Spoke from source.

    Supports installation from local directories. Git and registry sources
    are planned for future implementation.

    Args:
        source: Installation source - local path, git URL, or registry name
        editable: If True, symlink instead of copy (for development)
        name: Override Spoke name from manifest (optional)

    Returns:
        SpokeMetadata object for installed Spoke

    Example:
        ```bash
        # Install from local directory
        $ axium spoke install ~/my-spokes/aws

        # Install in editable mode (symlink)
        $ axium spoke install ~/dev/my-spoke --editable
        ```

    Raises:
        FileNotFoundError: If source doesn't exist or invalid
        ValueError: If spoke.yaml invalid
        FileExistsError: If Spoke already installed

    Note:
        - Validates spoke.yaml before installation
        - Adds entry to .spoke_metadata.json
        - Does not auto-reload (caller should call reload_spokes())
    """
    # Parse source type
    source_path = Path(source).expanduser().resolve()

    # TODO: Support git:// and registry:// sources in future
    if not source_path.exists():
        raise FileNotFoundError(f"Source path not found: {source}")

    if not source_path.is_dir():
        raise ValueError(f"Source must be a directory: {source}")

    # Validate Spoke structure
    manifest_data = validate_spoke(source_path)

    # Use provided name or manifest name
    spoke_name = name or manifest_data["name"]

    # Install (copy or symlink)
    dest_path, install_mode = _install_local(source_path, spoke_name, editable)

    # Create metadata entry
    metadata = SpokeMetadata(
        name=spoke_name,
        version=manifest_data["version"],
        description=manifest_data["description"],
        entrypoint=manifest_data["entrypoint"],
        source=f"local:{source_path}",
        install_mode=install_mode,
        installed_at=datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
        status="not-loaded",
    )

    # Save to metadata file
    metadata_data = _load_metadata_file()
    metadata_data["spokes"][spoke_name] = metadata.to_dict()
    _save_metadata_file(metadata_data)

    logger.info("Installed Spoke: %s (%s mode)", spoke_name, install_mode)
    print(f"✓ Spoke '{spoke_name}' installed at {dest_path}")

    return metadata

reload_spokes(spoke_name=None)

Reload spoke(s) dynamically without restarting daemon.

Unloads specified spoke(s) by clearing their commands from the registry, reloads their Python modules, and re-calls register() to re-register commands and event handlers.

Parameters:

Name	Type	Description	Default
spoke_name	str | None	Specific spoke to reload, or None to reload all	None

Returns:

Type	Description
list[str]	List of spoke names that were successfully reloaded

Raises:

Type	Description
ValueError	If specified spoke_name is not installed

Example

>>> reload_spokes("aws")  # Reload just aws spoke
['aws']
>>> reload_spokes()  # Reload all spokes
['aws', 'creds', 'k8s']

Note

• Updates last_loaded timestamp in metadata
• Sets status to "active" on success, "error" on failure
• Emits spoke_reloaded event for each successful reload
• Useful for development without daemon restarts

Source code in axium/core/spokes.py

def reload_spokes(spoke_name: str | None = None) -> list[str]:
    """
    Reload spoke(s) dynamically without restarting daemon.

    Unloads specified spoke(s) by clearing their commands from the registry,
    reloads their Python modules, and re-calls register() to re-register
    commands and event handlers.

    Args:
        spoke_name: Specific spoke to reload, or None to reload all

    Returns:
        List of spoke names that were successfully reloaded

    Raises:
        ValueError: If specified spoke_name is not installed

    Example:
        ```python
        >>> reload_spokes("aws")  # Reload just aws spoke
        ['aws']
        >>> reload_spokes()  # Reload all spokes
        ['aws', 'creds', 'k8s']
        ```

    Note:
        - Updates last_loaded timestamp in metadata
        - Sets status to "active" on success, "error" on failure
        - Emits spoke_reloaded event for each successful reload
        - Useful for development without daemon restarts
    """
    import importlib.util
    import sys

    metadata_data = _load_metadata_file()
    spokes_to_reload = {}

    if spoke_name:
        # Reload specific spoke
        if spoke_name not in metadata_data["spokes"]:
            raise ValueError(f"Spoke '{spoke_name}' is not installed")
        spokes_to_reload[spoke_name] = metadata_data["spokes"][spoke_name]
    else:
        # Reload all spokes
        spokes_to_reload = metadata_data["spokes"]

    if not spokes_to_reload:
        logger.info("No spokes to reload")
        return []

    # Import parent_app for re-registration
    # Note: In practice, this would be passed as a parameter or accessed from global state
    # For now, we'll defer this to the CLI layer which has access to the app
    from axium.core.cli import app as parent_app

    reloaded_names = []

    for name, spoke_meta in spokes_to_reload.items():
        spoke_dir = SPOKES_DIR / name

        if not spoke_dir.exists():
            logger.warning("Spoke directory not found: %s", spoke_dir)
            metadata_data["spokes"][name]["status"] = "error"
            continue

        try:
            # Step 1: Unload commands from registry
            _unload_spoke(name)

            # Step 2: Parse manifest
            spoke_yaml = spoke_dir / "spoke.yaml"
            manifest_data = yaml.safe_load(spoke_yaml.read_text())

            # Step 3: Add spoke dir to sys.path if not already there
            spoke_dir_str = str(spoke_dir)
            if spoke_dir_str not in sys.path:
                sys.path.insert(0, spoke_dir_str)

            # Step 4: Reload module
            _reload_spoke_module(spoke_dir, manifest_data)

            # Step 5: Snapshot Typer app commands BEFORE calling register
            from axium.core import registry

            # Get set of command names currently in the Typer app
            typer_commands_before = {
                cmd.name or cmd.callback.__name__
                for cmd in parent_app.registered_commands
            }

            # Step 6: Re-import and call register()
            entrypoint = manifest_data["entrypoint"]
            module_name, func_name = entrypoint.split(":")

            module = sys.modules[module_name]
            register_func = getattr(module, func_name)
            register_func(parent_app, get_event_bus())

            # Step 7: Find NEW commands added by this spoke
            typer_commands_after = {
                cmd.name or cmd.callback.__name__
                for cmd in parent_app.registered_commands
            }
            new_commands = typer_commands_after - typer_commands_before

            # Step 8: Only register the NEW commands in the registry with spoke source
            for cmd in parent_app.registered_commands:
                cmd_name = cmd.name or cmd.callback.__name__
                if cmd_name in new_commands:
                    help_text = cmd.help
                    if not help_text and cmd.callback.__doc__:
                        doc_lines = [
                            line.strip()
                            for line in cmd.callback.__doc__.split("\n")
                            if line.strip()
                        ]
                        help_text = doc_lines[0] if doc_lines else ""

                    registry.register_command(
                        name=cmd_name,
                        help=help_text or "",
                        source=name,
                        group=None,
                        callback=cmd.callback,
                    )

            # Step 9: Invalidate config cache
            from axium.core import config

            config.invalidate_cache(name)
            logger.debug("Invalidated config cache for spoke: %s", name)

            # Step 10: Load permissions into daemon
            try:
                from axium.core.ipc import send_request_sync

                resp = send_request_sync(
                    {
                        "cmd": "load_spoke_permissions",
                        "spoke": name,
                        "spoke_yaml_path": str(spoke_yaml),
                    }
                )
                if resp.get("ok"):
                    logger.debug("Reloaded permissions for spoke: %s", name)
            except Exception as e:
                logger.debug("Could not reload permissions for spoke %s: %s", name, e)

            # Step 11: Update metadata
            now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
            metadata_data["spokes"][name]["last_loaded"] = now
            metadata_data["spokes"][name]["status"] = "active"

            # Step 12: Emit event
            get_event_bus().emit("spoke_reloaded", name)

            logger.info("Reloaded Spoke: %s", name)
            reloaded_names.append(name)

        except Exception as e:
            logger.error("Failed to reload Spoke '%s': %s", name, e)
            metadata_data["spokes"][name]["status"] = "error"

    # Save updated metadata
    _save_metadata_file(metadata_data)

    logger.info(
        "Reloaded %d spoke(s): %s", len(reloaded_names), ", ".join(reloaded_names)
    )
    return reloaded_names

load_spokes(parent_app)

Discover and load all Spokes from ~/.config/axium/spokes/.

Scans the spokes directory for subdirectories containing spoke.yaml. For each valid Spoke: 1. Parse spoke.yaml manifest 2. Import the entrypoint module 3. Call register(app, events) with CLI app and EventBus 4. Emit spoke_loaded event

Parameters:

Name	Type	Description	Default
parent_app	Any	Typer application instance to pass to Spoke register functions	required

Spoke Discovery

• Searches ~/.config/axium/spokes/
• Each subdirectory is checked for spoke.yaml
• Manifest must contain 'name' and 'entrypoint'
• Entrypoint format: "spoke.module:function" (e.g., "aws.main:register")

Register Signature

def register(app: Typer, events: EventBus) -> None

Example

>>> from typer import Typer
>>> app = Typer()
>>> load_spokes(app)
# All Spokes in ~/.config/axium/spokes/ are now loaded

Note

• Spokes are loaded in alphabetical order by directory name
• Errors in individual Spokes are logged but don't stop loading
• The spoke directory is added to sys.path for imports
• Returns silently if spokes directory doesn't exist

Source code in axium/core/spokes.py

def load_spokes(parent_app: Any) -> None:
    """
    Discover and load all Spokes from ~/.config/axium/spokes/.

    Scans the spokes directory for subdirectories containing spoke.yaml.
    For each valid Spoke:
    1. Parse spoke.yaml manifest
    2. Import the entrypoint module
    3. Call register(app, events) with CLI app and EventBus
    4. Emit spoke_loaded event

    Args:
        parent_app: Typer application instance to pass to Spoke register functions

    Spoke Discovery:
        - Searches ~/.config/axium/spokes/
        - Each subdirectory is checked for spoke.yaml
        - Manifest must contain 'name' and 'entrypoint'
        - Entrypoint format: "spoke.module:function" (e.g., "aws.main:register")

    Register Signature:
        def register(app: Typer, events: EventBus) -> None

    Example:
        ```python
        >>> from typer import Typer
        >>> app = Typer()
        >>> load_spokes(app)
        # All Spokes in ~/.config/axium/spokes/ are now loaded
        ```

    Note:
        - Spokes are loaded in alphabetical order by directory name
        - Errors in individual Spokes are logged but don't stop loading
        - The spoke directory is added to sys.path for imports
        - Returns silently if spokes directory doesn't exist
    """
    if not SPOKES_DIR.exists():
        return

    # Load metadata to track load status
    metadata_data = _load_metadata_file()
    metadata_updated = False

    for spoke_dir in sorted(SPOKES_DIR.iterdir()):
        if not spoke_dir.is_dir():
            continue
        manifest = spoke_dir / "spoke.yaml"
        if not manifest.exists():
            continue

        spoke_name = spoke_dir.name

        try:
            data = yaml.safe_load(manifest.read_text())
            entry = data.get("entrypoint")
            if not entry:
                logger.warning("Spoke %s missing entrypoint", spoke_name)
                continue

            module_path, func_name = entry.split(":")

            # Check if this is a local spoke module (format: "spoke.main" or "spoke.module")
            # Spokes should use spoke-relative paths like "creds.main" not "main"
            if "." in module_path:
                # Extract file path (e.g., "creds.main" -> "main.py")
                module_parts = module_path.split(".")
                parent_module_name = module_parts[0]
                child_module_name = module_parts[-1]
                module_file = spoke_dir / f"{child_module_name}.py"

                # Create fake parent package if it doesn't exist
                if parent_module_name not in sys.modules:
                    import types

                    parent_module = types.ModuleType(parent_module_name)
                    parent_module.__path__ = [str(spoke_dir)]
                    parent_module.__file__ = str(spoke_dir / "__init__.py")
                    sys.modules[parent_module_name] = parent_module
                    logger.debug("Created parent module: %s", parent_module_name)
            else:
                # Legacy format: "main" -> "main.py"
                module_file = spoke_dir / f"{module_path}.py"

            if module_file.exists():
                # Load local module directly from file
                import importlib.util

                spec = importlib.util.spec_from_file_location(module_path, module_file)
                if spec is None or spec.loader is None:
                    raise ImportError(
                        f"Cannot load module {module_path} from {module_file}"
                    )

                module = importlib.util.module_from_spec(spec)
                sys.modules[module_path] = module
                spec.loader.exec_module(module)
                logger.debug(
                    "Loaded local spoke module: %s from %s", module_path, module_file
                )
            else:
                # Try standard import (for installed packages or .py suffix handling)
                if module_path.endswith(".py"):
                    sys.path.insert(0, str(spoke_dir))
                    module_path = Path(module_path).stem

                module = importlib.import_module(module_path)
                logger.debug("Imported spoke module: %s", module_path)

            register_func = getattr(module, func_name)

            # Snapshot Typer app commands before spoke registers
            from axium.core import registry

            typer_commands_before = {
                cmd.name or cmd.callback.__name__
                for cmd in parent_app.registered_commands
            }

            # Call register with (app, events) signature
            register_func(parent_app, _event_bus)

            # Find NEW commands added by this spoke
            typer_commands_after = {
                cmd.name or cmd.callback.__name__
                for cmd in parent_app.registered_commands
            }
            new_commands = typer_commands_after - typer_commands_before

            # Only register the NEW commands in the registry with spoke source
            for cmd in parent_app.registered_commands:
                cmd_name = cmd.name or cmd.callback.__name__
                if cmd_name in new_commands:
                    help_text = cmd.help
                    if not help_text and cmd.callback.__doc__:
                        doc_lines = [
                            line.strip()
                            for line in cmd.callback.__doc__.split("\n")
                            if line.strip()
                        ]
                        help_text = doc_lines[0] if doc_lines else ""

                    registry.register_command(
                        name=cmd_name,
                        help=help_text or "",
                        source=spoke_name,
                        group=None,
                        callback=cmd.callback,
                    )

            logger.info(
                "Loaded Spoke: %s (%d commands registered)",
                spoke_name,
                len(new_commands),
            )

            # Emit spoke_loaded event
            _event_bus.emit("spoke_loaded", spoke_name)

            # Load permissions into daemon (if daemon is running)
            try:
                from axium.core.ipc import send_request_sync

                resp = send_request_sync(
                    {
                        "cmd": "load_spoke_permissions",
                        "spoke": spoke_name,
                        "spoke_yaml_path": str(manifest),
                    }
                )
                if resp.get("ok"):
                    logger.debug("Loaded permissions for spoke: %s", spoke_name)
                else:
                    logger.warning(
                        "Failed to load permissions for spoke %s: %s",
                        spoke_name,
                        resp.get("error", "unknown error"),
                    )
            except Exception as e:
                # Daemon not running or IPC failed - non-fatal
                logger.debug(
                    "Could not load permissions for spoke %s (daemon may not be running): %s",
                    spoke_name,
                    e,
                )

            # Update metadata: set last_loaded and status=active
            if spoke_name in metadata_data["spokes"]:
                now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
                metadata_data["spokes"][spoke_name]["last_loaded"] = now
                metadata_data["spokes"][spoke_name]["status"] = "active"
                metadata_updated = True

        except Exception as e:
            logger.error("Failed to load spoke %s: %s", spoke_name, e)

            # Update metadata: set status=error
            if spoke_name in metadata_data["spokes"]:
                metadata_data["spokes"][spoke_name]["status"] = "error"
                metadata_updated = True

    # Save updated metadata if any changes were made
    if metadata_updated:
        _save_metadata_file(metadata_data)

list_spokes()

List all installed Spokes to stdout.

Scans ~/.config/axium/spokes/ and displays Spoke names with their directory paths. Validates spoke.yaml exists and parses the name.

Output Format

Installed Spokes: - aws (~/.config/axium/spokes/aws) - k8s (~/.config/axium/spokes/k8s) (none found) # if directory empty/missing

Example

>>> list_spokes()
Installed Spokes:
  - aws (~/.config/axium/spokes/aws)
  - k8s (~/.config/axium/spokes/k8s)

Note

• Displays "(none found)" if no Spokes are installed
• Shows "(invalid manifest: ...)" for Spokes with malformed YAML
• Does not attempt to load Spokes, only lists them

Source code in axium/core/spokes.py

def list_spokes() -> None:
    """
    List all installed Spokes to stdout.

    Scans ~/.config/axium/spokes/ and displays Spoke names with their
    directory paths. Validates spoke.yaml exists and parses the name.

    Output Format:
        Installed Spokes:
          - aws (~/.config/axium/spokes/aws)
          - k8s (~/.config/axium/spokes/k8s)
          (none found)  # if directory empty/missing

    Example:
        ```bash
        >>> list_spokes()
        Installed Spokes:
          - aws (~/.config/axium/spokes/aws)
          - k8s (~/.config/axium/spokes/k8s)
        ```

    Note:
        - Displays "(none found)" if no Spokes are installed
        - Shows "(invalid manifest: ...)" for Spokes with malformed YAML
        - Does not attempt to load Spokes, only lists them
    """
    print("Installed Spokes:")
    if not SPOKES_DIR.exists():
        print("  (none found)")
        return
    for spoke_dir in sorted(SPOKES_DIR.iterdir()):
        if not spoke_dir.is_dir():
            continue
        manifest = spoke_dir / "spoke.yaml"
        if not manifest.exists():
            continue
        try:
            data = yaml.safe_load(manifest.read_text())
            print(f"  - {data.get('name')} ({spoke_dir})")
        except Exception as e:
            print(f"  - {spoke_dir.name} (invalid manifest: {e})")