Hook System

Kollab CLI uses a comprehensive event-driven hook system that allows plugins to intercept and modify behavior at every stage of operation.

Overview

The hook system is built on an event bus architecture where:

Events are emitted at key points in the application lifecycle
Hooks are async callbacks that respond to specific event types
Priority determines the order hooks execute (higher values first)
Data flow allows hooks to modify and enrich data as it passes through

Event Types

User Input Events

Event	Description
`USER_INPUT_PRE`	Before processing user input
`USER_INPUT`	During user input processing
`USER_INPUT_POST`	After user input has been processed
`KEY_PRESS_PRE`	Before key press is handled
`KEY_PRESS`	During key press handling
`PASTE_DETECTED`	When clipboard paste is detected

LLM Events

Event	Description
`LLM_REQUEST_PRE`	Before sending request to LLM API
`LLM_REQUEST`	During LLM API call
`LLM_RESPONSE_PRE`	Before processing LLM response
`LLM_RESPONSE_POST`	After LLM response is processed
`LLM_THINKING`	During LLM thinking/reasoning phase
`CANCEL_REQUEST`	When LLM request is cancelled

Tool Events

Event	Description
`TOOL_CALL_PRE`	Before executing a tool call
`TOOL_CALL`	During tool execution
`TOOL_CALL_POST`	After tool execution completes

System Events

Event	Description
`SYSTEM_STARTUP`	Application startup
`SYSTEM_SHUTDOWN`	Application shutdown
`RENDER_FRAME`	Each render frame update

Slash Command Events

Event	Description
`SLASH_COMMAND_DETECTED`	When a slash command is detected
`SLASH_COMMAND_EXECUTE`	During command execution
`SLASH_COMMAND_COMPLETE`	After command completes
`SLASH_COMMAND_ERROR`	When command execution fails

Hook Priority

Hooks execute in priority order, with higher values executing first:

Priority	Value	Use Case
`SYSTEM`	1000	Core system operations
`SECURITY`	900	Security checks and validation
`PREPROCESSING`	500	Data transformation before processing
`LLM`	100	LLM-related processing
`POSTPROCESSING`	50	Data transformation after processing
`DISPLAY`	10	Display and rendering operations

Registering Hooks

from core.events import Event, EventType, Hook, HookPriority

async def register_hooks(self):
    """Register plugin hooks."""

    # Create a hook
    hook = Hook(
        name="my_input_processor",
        plugin_name=self.name,
        event_type=EventType.USER_INPUT_PRE,
        priority=HookPriority.PREPROCESSING.value,
        callback=self.process_input,
        timeout=30,           # Max execution time (seconds)
        retry_attempts=3,     # Retry on failure
        error_action="continue"  # or "stop"
    )

    # Register with event bus
    await self.event_bus.register_hook(hook)

Hook Callbacks

Hook callbacks are async functions that receive data and the event object:

async def process_input(self, data: dict, event: Event) -> dict:
    """Process user input.

    Args:
        data: Event-specific data (varies by event type)
        event: The event object with metadata

    Returns:
        Modified data dict to pass to next hook
    """
    message = data.get("message", "")

    # Enrich the data
    enriched = {
        "original_message": message,
        "message": message.strip(),
        "processed_at": event.timestamp,
        "word_count": len(message.split())
    }

    return enriched

The data returned by each hook is passed to the next hook in the chain. Hooks can modify, enrich, or filter data as needed.

Example: Security Hook

Here's a complete example of a security validation hook:

from core.events import EventType, Hook, HookPriority

class SecurityPlugin:
    """Plugin that validates tool calls for security."""

    HIGH_RISK_PATTERNS = ["rm -rf", "sudo", "chmod 777"]

    async def register_hooks(self):
        hook = Hook(
            name="security_tool_validator",
            plugin_name=self.name,
            event_type=EventType.TOOL_CALL_PRE,
            priority=HookPriority.SECURITY.value,
            callback=self.validate_tool_call
        )
        await self.event_bus.register_hook(hook)

    async def validate_tool_call(self, data, event):
        """Validate tool calls before execution."""
        command = data.get("command", "")

        # Check for high-risk patterns
        for pattern in self.HIGH_RISK_PATTERNS:
            if pattern in command.lower():
                return {
                    **data,
                    "blocked": True,
                    "risk_level": "high",
                    "reason": f"Blocked pattern: {pattern}"
                }

        return {**data, "validated": True, "risk_level": "low"}