Event System Reference¶

Axium provides a synchronous event bus for coordinating between core functionality and Spokes/Gears. Events are emitted at key lifecycle points, allowing extensions to react to state changes.

Event Bus Architecture¶

The EventBus uses a publish-subscribe pattern where:

Publishers: Axium core (daemon, CLI) emits events at specific lifecycle points
Subscribers: Spokes and Gears register callbacks to react to events
Synchronous: All callbacks execute in registration order before emit() returns
Error Handling: Exceptions in callbacks are logged but don't stop propagation

Standard Events¶

Environment Events¶

`env_change`¶

Emitted when the active environment changes.

Signature:

def on_env_change(new_env: str, old_env: str, pane: str | None = None):
    pass

Parameters: - new_env: Name of newly activated environment (or None if clearing) - old_env: Name of previously active environment (or None if none was set) - pane: (keyword) Optional tmux pane ID if change is pane-specific

Emitted By: - axium env set <name> - Global environment change - axium env set <name> --pane <id> - Pane-specific environment change - IPC set_env command - IPC set_pane_env command

When: - After environment state is updated in daemon - Before HUD cache is refreshed

Common Uses: - Refresh credentials when switching environments - Update HUD segments with environment-specific data - Clear caches tied to previous environment - Notify user of environment-specific warnings

Example:

def register(app, events):
    def on_env_change(new_env: str, old_env: str, **kwargs):
        pane = kwargs.get("pane")
        if pane:
            print(f"Pane {pane}: {old_env} → {new_env}")
        else:
            print(f"Global: {old_env} → {new_env}")

        # Check credentials for new environment
        check_credentials(new_env)

    events.on("env_change", on_env_change)

Spoke Lifecycle Events¶

`spoke_loaded`¶

Emitted when a Spoke finishes initial loading.

Signature:

def on_spoke_loaded(spoke_name: str):
    pass

Parameters: - spoke_name: Name of the Spoke that loaded

Emitted By: - Daemon during startup when loading all Spokes - axium spoke install <name> after successful installation

When: - After Spoke's register() function has been called - After Spoke is marked as "active" in metadata

Common Uses: - Initialize spoke state (e.g., check credentials on first load) - Set initial HUD segments - One-time setup that shouldn't run on reload

Example:

from axium.core.hud import get_registry

def register(app, events):
    def on_spoke_loaded(spoke_name: str):
        if spoke_name == "creds":
            # Perform initial credential check and update cached HUD
            from axium.core import api
            registry = get_registry()
            env_name = api.get_active_env()
            context = {"env": env_name}
            registry.update_cached_segments(context)

    events.on("spoke_loaded", on_spoke_loaded)

`spoke_reloaded`¶

Emitted when a Spoke is reloaded.

Signature:

def on_spoke_reloaded(spoke_name: str):
    pass

Parameters: - spoke_name: Name of the Spoke that reloaded

Emitted By: - axium daemon reload after reloading all Spokes - axium spoke reload <name> after reloading specific Spoke

When: - After Spoke's register() function has been re-called - After Spoke metadata is updated

Common Uses: - Re-initialize state that was lost during reload - Refresh configuration from files - Clear stale caches

Example:

def register(app, events):
    def on_spoke_reloaded(spoke_name: str):
        if spoke_name == "aws":
            # Reload AWS configuration
            reload_aws_config()

    events.on("spoke_reloaded", on_spoke_reloaded)

`spoke_unloaded`¶

Emitted when a Spoke is unloaded.

Signature:

def on_spoke_unloaded(spoke_name: str):
    pass

Parameters: - spoke_name: Name of the Spoke that unloaded

Emitted By: - axium spoke uninstall <name> after removing Spoke - Daemon shutdown (for all loaded Spokes)

When: - After Spoke has been removed from active registry - Before Spoke files are deleted (on uninstall)

Common Uses: - Cleanup resources - Save state to disk - Remove HUD segments

Example:

def register(app, events):
    def on_spoke_unloaded(spoke_name: str):
        if spoke_name == "creds":
            # Clear HUD segment
            api.update_hud_segment("creds", "")

    events.on("spoke_unloaded", on_spoke_unloaded)

Gear Lifecycle Events¶

`gear_loaded`¶

Emitted when a Gear finishes loading.

Signature:

def on_gear_loaded(gear_name: str):
    pass

Parameters: - gear_name: Name of the Gear that loaded

Emitted By: - Daemon during startup when loading all Gears - axium gear install <name> after successful installation

When: - After Gear's register function has been called - After Gear is marked as "active" in metadata

Common Uses: - Initialize gear-specific state - Register gear-provided commands in tmux - Verify gear permissions

`gear_unloaded`¶

Emitted when a Gear is unloaded.

Signature:

def on_gear_unloaded(gear_name: str):
    pass

Parameters: - gear_name: Name of the Gear that unloaded

Emitted By: - axium gear uninstall <name> after removing Gear - Daemon shutdown

When: - After Gear has been removed from active registry

Common Uses: - Cleanup gear resources - Remove tmux key bindings

Daemon Events¶

`daemon_reload`¶

Emitted when daemon configuration is reloaded.

Signature:

def on_daemon_reload():
    pass

Parameters: None

Emitted By: - axium daemon reload command - IPC reload command

When: - After all Spokes have been reloaded - After configuration files (envs.yaml, prefixes.yaml, etc.) are re-read - After HUD cache is cleared

Common Uses: - Refresh spoke-specific configuration - Clear caches that depend on config files - Re-validate setup

Example:

def register(app, events):
    def on_daemon_reload():
        # Reload our configuration
        global config
        config = load_config()

    events.on("daemon_reload", on_daemon_reload)

`config_reloaded`¶

Emitted when all spoke configurations are cleared and reloaded.

Signature:

def on_config_reloaded():
    pass

Parameters: None

Emitted By: - axium daemon reload command - Configuration file changes (if file watching is enabled)

When: - After all cached spoke configs are cleared - After config files are re-read from disk

Common Uses: - Reload spoke-specific configuration from ~/.config/axium/overrides/ - Validate new configuration values - Update state based on config changes

HUD Events¶

`hud_segment_updated`¶

Emitted when a HUD segment value changes.

Signature:

def on_hud_segment_updated(spoke: str, value: str):
    pass

Parameters: - spoke: Name of the spoke that updated its segment - value: New segment value (empty string if cleared)

Emitted By: - api.update_hud_segment() calls from Spokes - IPC update_hud_segment command

When: - After segment value is stored in daemon - After HUD cache is refreshed for all panes

Common Uses: - Track when other spokes update their HUD - Coordinate related HUD segments - Debug HUD rendering issues

Example:

def register(app, events):
    def on_hud_updated(spoke: str, value: str):
        logger.debug(f"HUD segment updated: {spoke} = {value}")

    events.on("hud_segment_updated", on_hud_updated)

`hud_refresh`¶

Emitted when HUD should be regenerated.

Signature:

def on_hud_refresh():
    pass

Parameters: None

Emitted By: - axium daemon reload - Environment changes - HUD segment updates

When: - Before HUD cache is regenerated for all panes

Common Uses: - Trigger expensive HUD calculations only when needed - Update dynamic HUD segments - Clear HUD-related caches

Event Flow Diagrams¶

Environment Change Flow¶

User runs: axium env set prod
    ↓
CLI → IPC set_env {"env": "prod"}
    ↓
Daemon updates state: active_env = "prod"
    ↓
Daemon emits: env_change("prod", "dev")
    ↓
Spokes receive callbacks (in registration order)
    ↓
Daemon refreshes HUD cache for all panes
    ↓
CLI exits silently (success)

Daemon Reload Flow¶

User runs: axium daemon reload
    ↓
CLI → IPC reload {}
    ↓
Daemon reloads config files
    ↓
Daemon emits: hud_refresh
    ↓
Daemon reloads all Spokes
    ↓
Daemon emits: spoke_reloaded for each Spoke
    ↓
Daemon emits: daemon_reload
    ↓
Daemon emits: config_reloaded
    ↓
Daemon refreshes HUD cache
    ↓
CLI exits silently (success)

Spoke Installation Flow¶

User runs: axium spoke install myspoke
    ↓
CLI copies/symlinks spoke to ~/.config/axium/spokes/myspoke
    ↓
CLI creates metadata entry
    ↓
CLI → IPC reload {} (triggers daemon reload)
    ↓
Daemon discovers new spoke
    ↓
Daemon calls myspoke.register(app, events)
    ↓
Daemon emits: spoke_loaded("myspoke")
    ↓
Other spokes receive callbacks
    ↓
CLI outputs: "Spoke 'myspoke' installed successfully"

Best Practices¶

Event Handler Guidelines¶

Keep handlers fast: Events are synchronous, slow handlers block daemon
Handle errors gracefully: Exceptions are logged but don't stop propagation
Use keyword arguments: Event signatures may gain parameters in future
Avoid recursion: Don't emit events from within handlers for same event
Log appropriately: Use logger.debug() for routine events, logger.info() for important changes

Example: Robust Event Handler¶

import logging

logger = logging.getLogger("myspoke")

def register(app, events):
    def on_env_change(new_env: str, old_env: str, **kwargs):
        """Handle environment change with proper error handling."""
        try:
            # Fast operations only
            pane = kwargs.get("pane")

            # Check if this is relevant to us
            if not new_env:
                logger.debug("Environment cleared")
                return

            # Perform lightweight updates
            update_hud_for_env(new_env)

            # Defer expensive operations
            if needs_heavy_work(new_env):
                # Use daemon_exec for background work
                from axium.core import api
                api.daemon_exec("myspoke", f"myspoke refresh --env {new_env}")

        except Exception as e:
            # Log but don't crash
            logger.error(f"Failed to handle env_change: {e}")

    events.on("env_change", on_env_change)

Performance Considerations¶

Environment changes: May happen frequently in tmux (pane switching)
HUD refresh: Triggers on every status line update (~1-5 seconds in tmux)
Daemon reload: Infrequent, can do heavier work

Choose the right event for your use case: - Fast checks (<10ms): Use events directly - Medium checks (10-100ms): Use events but cache results - Slow checks (>100ms): Use daemon_exec() for background processing

Custom Events¶

Spokes can emit custom events for coordination:

def register(app, events):
    # Emit custom event
    events.emit("myspoke_ready", version="1.0.0")

    # Listen to custom events from other spokes
    def on_other_spoke_event(data):
        print(f"Received: {data}")

    events.on("other_spoke_ready", on_other_spoke_event)

Note: Custom events are not guaranteed to persist across daemon versions. Use for inter-spoke coordination only, not for core functionality.

Event System Reference¶

Event Bus Architecture¶

Standard Events¶

Environment Events¶

env_change¶

Spoke Lifecycle Events¶

spoke_loaded¶

spoke_reloaded¶

spoke_unloaded¶

Gear Lifecycle Events¶

gear_loaded¶

gear_unloaded¶

Daemon Events¶

daemon_reload¶

config_reloaded¶

HUD Events¶

hud_segment_updated¶

hud_refresh¶

Event Flow Diagrams¶

Environment Change Flow¶

Daemon Reload Flow¶

Spoke Installation Flow¶

Best Practices¶

Event Handler Guidelines¶

Example: Robust Event Handler¶

Performance Considerations¶

Custom Events¶

See Also¶

`env_change`¶

`spoke_loaded`¶

`spoke_reloaded`¶

`spoke_unloaded`¶

`gear_loaded`¶

`gear_unloaded`¶

`daemon_reload`¶

`config_reloaded`¶

`hud_segment_updated`¶

`hud_refresh`¶