Hooks

Contract

HookProvider

Bases: ABC

Abstract base class for hook providers.

Hook providers implement lifecycle callbacks that are invoked during agent execution. Multiple providers can be registered, with execution order determined by priority (lower = earlier).

Example

class MyLoggingHook(HookProvider): @property def priority(self) -> int: return HookPriority.OBSERVABILITY_DEFAULT

async def on_before_invocation(
    self, prompt: str, state: AgentState
) -> AgentState:
    print(f"Starting: {prompt[:50]}...")
    return state

async def on_after_invocation(
    self, state: AgentState, success: bool
) -> None:
    print(f"Completed: success={success}")

priority abstractmethod property

priority: int

Hook priority (lower = earlier execution).

Use HookPriority constants for standard ranges.

name property

name: str

Hook provider name for identification.

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Called before agent starts processing.

Parameters:

Name	Type	Description	Default
prompt	str	The user prompt being processed	required
state	AgentState	Current agent state	required

Returns:

Type	Description
AgentState	Potentially modified agent state

Source code in src/locus/hooks/provider.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Called before agent starts processing.

    Args:
        prompt: The user prompt being processed
        state: Current agent state

    Returns:
        Potentially modified agent state
    """
    return state

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

Called after agent completes processing.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution completed successfully	required

Source code in src/locus/hooks/provider.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Called after agent completes processing.

    Args:
        state: Final agent state
        success: Whether execution completed successfully
    """

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent) -> None

Called before tool execution.

Modify event.arguments to change tool inputs. Set event.cancel = True or a string reason to skip execution. event.tool_name and event.tool_call_id are read-only.

Parameters:

Name	Type	Description	Default
event	BeforeToolCallEvent	Write-protected event. Writable: arguments, cancel.	required

Source code in src/locus/hooks/provider.py

async def on_before_tool_call(
    self,
    event: BeforeToolCallEvent,
) -> None:
    """Called before tool execution.

    Modify event.arguments to change tool inputs.
    Set event.cancel = True or a string reason to skip execution.
    event.tool_name and event.tool_call_id are read-only.

    Args:
        event: Write-protected event. Writable: arguments, cancel.
    """

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

Called after tool execution.

Set event.retry = True to re-execute the tool. Set event.result to replace the tool result. event.tool_name and event.error are read-only.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event. Writable: result, retry.	required

Source code in src/locus/hooks/provider.py

async def on_after_tool_call(
    self,
    event: AfterToolCallEvent,
) -> None:
    """Called after tool execution.

    Set event.retry = True to re-execute the tool.
    Set event.result to replace the tool result.
    event.tool_name and event.error are read-only.

    Args:
        event: Write-protected event. Writable: result, retry.
    """

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Called at the start of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in src/locus/hooks/provider.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the start of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Called at the end of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in src/locus/hooks/provider.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the end of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_before_model_call async

on_before_model_call(event: BeforeModelCallEvent) -> None

Called before each model.complete() call.

Modify event.messages to change what the model sees. event.tools is read-only (inspect only).

Parameters:

Name	Type	Description	Default
event	BeforeModelCallEvent	Write-protected event. Writable: messages.	required

Source code in src/locus/hooks/provider.py

async def on_before_model_call(
    self,
    event: BeforeModelCallEvent,
) -> None:
    """Called before each model.complete() call.

    Modify event.messages to change what the model sees.
    event.tools is read-only (inspect only).

    Args:
        event: Write-protected event. Writable: messages.
    """

on_after_model_call async

on_after_model_call(event: AfterModelCallEvent) -> None

Called after each model.complete() call.

Set event.retry = True to discard response and re-call. Set event.response to replace the response. event.messages is read-only.

Parameters:

Name	Type	Description	Default
event	AfterModelCallEvent	Write-protected event. Writable: response, retry.	required

Source code in src/locus/hooks/provider.py

async def on_after_model_call(
    self,
    event: AfterModelCallEvent,
) -> None:
    """Called after each model.complete() call.

    Set event.retry = True to discard response and re-call.
    Set event.response to replace the response.
    event.messages is read-only.

    Args:
        event: Write-protected event. Writable: response, retry.
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in src/locus/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }

HookRegistry

HookRegistry()

Registry for managing hook providers.

The registry maintains a priority-ordered list of hook providers and dispatches lifecycle events to them in order.

Example

registry = HookRegistry() registry.add_provider(LoggingHook()) registry.add_provider(GuardrailsHook())

During agent execution

state = await registry.emit_before_invocation(prompt, state)

... agent runs ...

await registry.emit_after_invocation(state, success=True)

Initialize empty hook registry.

Source code in src/locus/hooks/registry.py

def __init__(self) -> None:
    """Initialize empty hook registry."""
    self._providers: list[HookProvider] = []
    self._sorted = True

providers property

providers: list[HookProvider]

Get all registered providers in priority order.

add_provider

add_provider(provider: HookProvider) -> None

Register a hook provider.

Parameters:

Name	Type	Description	Default
provider	HookProvider	Hook provider to register	required

Raises:

Type	Description
ValueError	If provider with same name already registered

Source code in src/locus/hooks/registry.py

def add_provider(self, provider: HookProvider) -> None:
    """Register a hook provider.

    Args:
        provider: Hook provider to register

    Raises:
        ValueError: If provider with same name already registered
    """
    for existing in self._providers:
        if existing.name == provider.name:
            msg = f"Hook provider '{provider.name}' already registered"
            raise ValueError(msg)

    self._providers.append(provider)
    self._sorted = False
    logger.debug(
        "Registered hook provider '%s' with priority %d",
        provider.name,
        provider.priority,
    )

remove_provider

remove_provider(name: str) -> bool

Remove a hook provider by name.

Parameters:

Name	Type	Description	Default
name	str	Name of the provider to remove	required

Returns:

Type	Description
bool	True if provider was removed, False if not found

Source code in src/locus/hooks/registry.py

def remove_provider(self, name: str) -> bool:
    """Remove a hook provider by name.

    Args:
        name: Name of the provider to remove

    Returns:
        True if provider was removed, False if not found
    """
    for i, provider in enumerate(self._providers):
        if provider.name == name:
            self._providers.pop(i)
            logger.debug("Removed hook provider '%s'", name)
            return True
    return False

get_provider

get_provider(name: str) -> HookProvider | None

Get a hook provider by name.

Parameters:

Name	Type	Description	Default
name	str	Name of the provider to find	required

Returns:

Type	Description
HookProvider | None	The provider if found, None otherwise

Source code in src/locus/hooks/registry.py

def get_provider(self, name: str) -> HookProvider | None:
    """Get a hook provider by name.

    Args:
        name: Name of the provider to find

    Returns:
        The provider if found, None otherwise
    """
    for provider in self._providers:
        if provider.name == name:
            return provider
    return None

__len__

__len__() -> int

Return number of registered providers.

Source code in src/locus/hooks/registry.py

def __len__(self) -> int:
    """Return number of registered providers."""
    return len(self._providers)

__contains__

__contains__(name: str) -> bool

Check if a provider with given name is registered.

Source code in src/locus/hooks/registry.py

def __contains__(self, name: str) -> bool:
    """Check if a provider with given name is registered."""
    return any(p.name == name for p in self._providers)

emit_before_invocation async

emit_before_invocation(prompt: str, state: AgentState) -> AgentState

Emit before_invocation event to all providers.

Parameters:

Name	Type	Description	Default
prompt	str	User prompt being processed	required
state	AgentState	Current agent state	required

Returns:

Type	Description
AgentState	Potentially modified agent state

Source code in src/locus/hooks/registry.py

async def emit_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Emit before_invocation event to all providers.

    Args:
        prompt: User prompt being processed
        state: Current agent state

    Returns:
        Potentially modified agent state
    """
    self._ensure_sorted()
    for provider in self._providers:
        try:
            state = await provider.on_before_invocation(prompt, state)
        except Exception:
            logger.exception(
                "Error in hook provider '%s' on_before_invocation",
                provider.name,
            )
            raise
    return state

emit_after_invocation async

emit_after_invocation(state: AgentState, success: bool) -> None

Emit after_invocation event to all providers.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution completed successfully	required

Source code in src/locus/hooks/registry.py

async def emit_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Emit after_invocation event to all providers.

    Args:
        state: Final agent state
        success: Whether execution completed successfully
    """
    self._ensure_sorted()
    errors: list[tuple[str, Exception]] = []
    # Reverse order: last-registered-first for proper teardown
    for provider in reversed(self._providers):
        try:
            await provider.on_after_invocation(state, success)
        except Exception as e:
            logger.exception(
                "Error in hook provider '%s' on_after_invocation",
                provider.name,
            )
            errors.append((provider.name, e))

    # Re-raise first error if any occurred
    if errors:
        name, error = errors[0]
        msg = f"Hook provider '{name}' failed in on_after_invocation: {error}"
        raise RuntimeError(msg) from error

emit_before_tool_call async

emit_before_tool_call(tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]

Emit before_tool_call event to all providers.

Parameters:

Name	Type	Description	Default
tool_name	str	Name of the tool being called	required
arguments	dict[str, Any]	Tool arguments	required

Returns:

Type	Description
dict[str, Any]	Potentially modified arguments

Source code in src/locus/hooks/registry.py

async def emit_before_tool_call(
    self,
    tool_name: str,
    arguments: dict[str, Any],
) -> dict[str, Any]:
    """Emit before_tool_call event to all providers.

    Args:
        tool_name: Name of the tool being called
        arguments: Tool arguments

    Returns:
        Potentially modified arguments
    """
    from locus.hooks.provider import BeforeToolCallEvent

    self._ensure_sorted()
    event = BeforeToolCallEvent(tool_name=tool_name, tool_call_id="", arguments=arguments)
    for provider in self._providers:
        try:
            await provider.on_before_tool_call(event)
        except Exception:
            logger.exception(
                "Error in hook provider '%s' on_before_tool_call",
                provider.name,
            )
            raise
    modified_arguments: dict[str, Any] = event.arguments
    return modified_arguments

emit_after_tool_call async

emit_after_tool_call(tool_name: str, result: Any, error: str | None, *, tool_call_id: str = '', arguments: dict[str, Any] | None = None) -> None

Emit after_tool_call event to all providers.

Parameters:

Name	Type	Description	Default
tool_name	str	Name of the tool that was called	required
result	Any	Tool result (if successful)	required
error	str | None	Error message (if failed)	required
tool_call_id	str	ID of the tool call (correlates with BeforeToolCallEvent).	''
arguments	dict[str, Any] | None	Arguments the tool was invoked with (post-hook mutation).	None

Source code in src/locus/hooks/registry.py

async def emit_after_tool_call(
    self,
    tool_name: str,
    result: Any,
    error: str | None,
    *,
    tool_call_id: str = "",
    arguments: dict[str, Any] | None = None,
) -> None:
    """Emit after_tool_call event to all providers.

    Args:
        tool_name: Name of the tool that was called
        result: Tool result (if successful)
        error: Error message (if failed)
        tool_call_id: ID of the tool call (correlates with BeforeToolCallEvent).
        arguments: Arguments the tool was invoked with (post-hook mutation).
    """
    from locus.hooks.provider import AfterToolCallEvent

    self._ensure_sorted()
    event = AfterToolCallEvent(
        tool_name=tool_name,
        result=result,
        error=error,
        tool_call_id=tool_call_id,
        arguments=arguments,
    )
    errors: list[tuple[str, Exception]] = []
    # Reverse order for proper teardown
    for provider in reversed(self._providers):
        try:
            await provider.on_after_tool_call(event)
        except Exception as e:
            logger.exception(
                "Error in hook provider '%s' on_after_tool_call",
                provider.name,
            )
            errors.append((provider.name, e))

    if errors:
        name, error_exc = errors[0]
        msg = f"Hook provider '{name}' failed in on_after_tool_call: {error_exc}"
        raise RuntimeError(msg) from error_exc

emit_iteration_start async

emit_iteration_start(iteration: int, state: AgentState) -> None

Emit iteration_start event to all providers.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number	required
state	AgentState	Current agent state	required

Source code in src/locus/hooks/registry.py

async def emit_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Emit iteration_start event to all providers.

    Args:
        iteration: Current iteration number
        state: Current agent state
    """
    self._ensure_sorted()
    tasks = [provider.on_iteration_start(iteration, state) for provider in self._providers]
    if tasks:
        await asyncio.gather(*tasks, return_exceptions=True)

emit_iteration_end async

emit_iteration_end(iteration: int, state: AgentState) -> None

Emit iteration_end event to all providers.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number	required
state	AgentState	Current agent state	required

Source code in src/locus/hooks/registry.py

async def emit_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Emit iteration_end event to all providers.

    Args:
        iteration: Current iteration number
        state: Current agent state
    """
    self._ensure_sorted()
    tasks = [provider.on_iteration_end(iteration, state) for provider in self._providers]
    if tasks:
        await asyncio.gather(*tasks, return_exceptions=True)

emit_before_model_call async

emit_before_model_call(messages: list[Any], tools: list[dict[str, Any]] | None) -> list[Any]

Emit before_model_call event to all providers.

Parameters:

Name	Type	Description	Default
messages	list[Any]	Messages about to be sent to the model	required
tools	list[dict[str, Any]] | None	Tool schemas (if any)	required

Returns:

Type	Description
list[Any]	Potentially modified messages list

Source code in src/locus/hooks/registry.py

async def emit_before_model_call(
    self,
    messages: list[Any],
    tools: list[dict[str, Any]] | None,
) -> list[Any]:
    """Emit before_model_call event to all providers.

    Args:
        messages: Messages about to be sent to the model
        tools: Tool schemas (if any)

    Returns:
        Potentially modified messages list
    """
    from locus.hooks.provider import BeforeModelCallEvent

    self._ensure_sorted()
    event = BeforeModelCallEvent(messages=messages, tools=tools)
    for provider in self._providers:
        try:
            await provider.on_before_model_call(event)
        except Exception:
            logger.exception(
                "Error in hook provider '%s' on_before_model_call",
                provider.name,
            )
            raise
    messages_out: list[Any] = event.messages
    return messages_out

emit_after_model_call async

emit_after_model_call(response: Any, messages: list[Any]) -> Any

Emit after_model_call event to all providers.

Parameters:

Name	Type	Description	Default
response	Any	The ModelResponse from the model	required
messages	list[Any]	The messages that were sent	required

Returns:

Type	Description
Any	Potentially modified response

Source code in src/locus/hooks/registry.py

async def emit_after_model_call(
    self,
    response: Any,
    messages: list[Any],
) -> Any:
    """Emit after_model_call event to all providers.

    Args:
        response: The ModelResponse from the model
        messages: The messages that were sent

    Returns:
        Potentially modified response
    """
    from locus.hooks.provider import AfterModelCallEvent

    self._ensure_sorted()
    event = AfterModelCallEvent(response=response, messages=messages)
    # Reverse order for proper teardown
    for provider in reversed(self._providers):
        try:
            await provider.on_after_model_call(event)
        except Exception:
            logger.exception(
                "Error in hook provider '%s' on_after_model_call",
                provider.name,
            )
            raise
    return event.response

emit async

emit(event_name: str, *args: Any, **kwargs: Any) -> Any

Generic event emission for custom hook points.

Parameters:

Name	Type	Description	Default
event_name	str	Name of the hook method to call	required
*args	Any	Positional arguments to pass	()
**kwargs	Any	Keyword arguments to pass	{}

Returns:

Type	Description
Any	Result from the last provider that returned a non-None value

Source code in src/locus/hooks/registry.py

async def emit(
    self,
    event_name: str,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Generic event emission for custom hook points.

    Args:
        event_name: Name of the hook method to call
        *args: Positional arguments to pass
        **kwargs: Keyword arguments to pass

    Returns:
        Result from the last provider that returned a non-None value
    """
    self._ensure_sorted()
    result = None
    for provider in self._providers:
        method = getattr(provider, event_name, None)
        if method is not None and callable(method):
            try:
                ret = await method(*args, **kwargs)
                if ret is not None:
                    result = ret
            except Exception:
                logger.exception(
                    "Error in hook provider '%s' %s",
                    provider.name,
                    event_name,
                )
                raise
    return result

Built-in hooks

LoggingHook

LoggingHook(level: int = logging.INFO, logger_name: str = 'locus.agent', extra: dict[str, Any] | None = None, log_arguments: bool = False, log_results: bool = False, priority: int = HookPriority.OBSERVABILITY_DEFAULT)

Bases: HookProvider

Hook provider that logs all lifecycle events.

Provides structured logging for agent execution with configurable log levels and optional extra context.

Example

Basic usage

registry.add_provider(LoggingHook())

With custom log level

registry.add_provider(LoggingHook(level=logging.DEBUG))

With structured context

registry.add_provider(LoggingHook( extra={"environment": "production", "service": "my-agent"} ))

Initialize logging hook.

Parameters:

Name	Type	Description	Default
level	int	Logging level (default: INFO)	INFO
logger_name	str	Name for the logger	'locus.agent'
extra	dict[str, Any] | None	Extra context to include in all log records	None
log_arguments	bool	Whether to log tool arguments (may contain sensitive data)	False
log_results	bool	Whether to log tool results (may be verbose)	False
priority	int	Hook priority	OBSERVABILITY_DEFAULT

Source code in src/locus/hooks/builtin/logging.py

def __init__(
    self,
    level: int = logging.INFO,
    logger_name: str = "locus.agent",
    extra: dict[str, Any] | None = None,
    log_arguments: bool = False,
    log_results: bool = False,
    priority: int = HookPriority.OBSERVABILITY_DEFAULT,
) -> None:
    """Initialize logging hook.

    Args:
        level: Logging level (default: INFO)
        logger_name: Name for the logger
        extra: Extra context to include in all log records
        log_arguments: Whether to log tool arguments (may contain sensitive data)
        log_results: Whether to log tool results (may be verbose)
        priority: Hook priority
    """
    self._level = level
    self._logger = logging.getLogger(logger_name)
    self._extra = extra or {}
    self._log_arguments = log_arguments
    self._log_results = log_results
    self._priority = priority

priority property

priority: int

Return hook priority.

name property

name: str

Return hook name.

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Log invocation start.

Parameters:

Name	Type	Description	Default
prompt	str	User prompt	required
state	AgentState	Agent state	required

Returns:

Type	Description
AgentState	Unchanged state

Source code in src/locus/hooks/builtin/logging.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Log invocation start.

    Args:
        prompt: User prompt
        state: Agent state

    Returns:
        Unchanged state
    """
    prompt_preview = prompt[:100] + "..." if len(prompt) > 100 else prompt
    self._log(
        "Agent invocation starting",
        run_id=state.run_id,
        agent_id=state.agent_id,
        prompt_preview=prompt_preview,
        prompt_length=len(prompt),
    )
    return state

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

Log invocation completion.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution succeeded	required

Source code in src/locus/hooks/builtin/logging.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Log invocation completion.

    Args:
        state: Final agent state
        success: Whether execution succeeded
    """
    duration_ms = (state.updated_at - state.started_at).total_seconds() * 1000
    self._log(
        "Agent invocation completed",
        run_id=state.run_id,
        agent_id=state.agent_id,
        success=success,
        iterations=state.iteration,
        confidence=state.confidence,
        tool_calls=len(state.tool_executions),
        errors=len(state.errors),
        duration_ms=duration_ms,
    )

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent) -> None

Log tool call start.

Parameters:

Name	Type	Description	Default
event	BeforeToolCallEvent	Write-protected event carrying tool_name and arguments. The hook only inspects them.	required

Source code in src/locus/hooks/builtin/logging.py

async def on_before_tool_call(self, event: BeforeToolCallEvent) -> None:
    """Log tool call start.

    Args:
        event: Write-protected event carrying ``tool_name`` and
            ``arguments``. The hook only inspects them.
    """
    log_data: dict[str, Any] = {"tool_name": event.tool_name}
    if self._log_arguments:
        log_data["arguments"] = event.arguments
    else:
        log_data["argument_keys"] = list(event.arguments.keys())

    self._log("Tool call starting", **log_data)

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

Log tool call completion.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event carrying tool_name, result, and error. The hook only inspects them.	required

Source code in src/locus/hooks/builtin/logging.py

async def on_after_tool_call(self, event: AfterToolCallEvent) -> None:
    """Log tool call completion.

    Args:
        event: Write-protected event carrying ``tool_name``,
            ``result``, and ``error``. The hook only inspects them.
    """
    log_data: dict[str, Any] = {
        "tool_name": event.tool_name,
        "success": event.error is None,
    }

    if event.error:
        log_data["error"] = event.error
    elif self._log_results and event.result is not None:
        result_str = str(event.result)
        log_data["result_preview"] = (
            result_str[:200] + "..." if len(result_str) > 200 else result_str
        )
        log_data["result_length"] = len(result_str)

    self._log("Tool call completed", **log_data)

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Log iteration start.

Parameters:

Name	Type	Description	Default
iteration	int	Iteration number	required
state	AgentState	Current state	required

Source code in src/locus/hooks/builtin/logging.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Log iteration start.

    Args:
        iteration: Iteration number
        state: Current state
    """
    self._log(
        "Iteration starting",
        run_id=state.run_id,
        iteration=iteration,
        max_iterations=state.max_iterations,
        confidence=state.confidence,
    )

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Log iteration end.

Parameters:

Name	Type	Description	Default
iteration	int	Iteration number	required
state	AgentState	Current state	required

Source code in src/locus/hooks/builtin/logging.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Log iteration end.

    Args:
        iteration: Iteration number
        state: Current state
    """
    self._log(
        "Iteration completed",
        run_id=state.run_id,
        iteration=iteration,
        confidence=state.confidence,
        messages=len(state.messages),
    )

on_before_model_call async

on_before_model_call(event: BeforeModelCallEvent) -> None

Called before each model.complete() call.

Modify event.messages to change what the model sees. event.tools is read-only (inspect only).

Parameters:

Name	Type	Description	Default
event	BeforeModelCallEvent	Write-protected event. Writable: messages.	required

Source code in src/locus/hooks/provider.py

async def on_before_model_call(
    self,
    event: BeforeModelCallEvent,
) -> None:
    """Called before each model.complete() call.

    Modify event.messages to change what the model sees.
    event.tools is read-only (inspect only).

    Args:
        event: Write-protected event. Writable: messages.
    """

on_after_model_call async

on_after_model_call(event: AfterModelCallEvent) -> None

Called after each model.complete() call.

Set event.retry = True to discard response and re-call. Set event.response to replace the response. event.messages is read-only.

Parameters:

Name	Type	Description	Default
event	AfterModelCallEvent	Write-protected event. Writable: response, retry.	required

Source code in src/locus/hooks/provider.py

async def on_after_model_call(
    self,
    event: AfterModelCallEvent,
) -> None:
    """Called after each model.complete() call.

    Set event.retry = True to discard response and re-call.
    Set event.response to replace the response.
    event.messages is read-only.

    Args:
        event: Write-protected event. Writable: response, retry.
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in src/locus/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }

StructuredLoggingHook

StructuredLoggingHook(level: int = logging.INFO, logger_name: str = 'locus.agent.structured', extra: dict[str, Any] | None = None, include_timestamps: bool = True, priority: int = HookPriority.OBSERVABILITY_DEFAULT)

Bases: LoggingHook

Logging hook with JSON-structured output.

Extends LoggingHook to emit structured JSON logs suitable for log aggregation systems like ELK, Datadog, or CloudWatch.

Example

import json import logging

Configure JSON handler

handler = logging.StreamHandler() handler.setFormatter(JsonFormatter()) logging.getLogger("locus.agent").addHandler(handler)

registry.add_provider(StructuredLoggingHook())

Initialize structured logging hook.

Parameters:

Name	Type	Description	Default
level	int	Logging level	INFO
logger_name	str	Logger name	'locus.agent.structured'
extra	dict[str, Any] | None	Extra context for all logs	None
include_timestamps	bool	Whether to include ISO timestamps	True
priority	int	Hook priority	OBSERVABILITY_DEFAULT

Source code in src/locus/hooks/builtin/logging.py

def __init__(
    self,
    level: int = logging.INFO,
    logger_name: str = "locus.agent.structured",
    extra: dict[str, Any] | None = None,
    include_timestamps: bool = True,
    priority: int = HookPriority.OBSERVABILITY_DEFAULT,
) -> None:
    """Initialize structured logging hook.

    Args:
        level: Logging level
        logger_name: Logger name
        extra: Extra context for all logs
        include_timestamps: Whether to include ISO timestamps
        priority: Hook priority
    """
    super().__init__(
        level=level,
        logger_name=logger_name,
        extra=extra,
        log_arguments=False,
        log_results=False,
        priority=priority,
    )
    self._include_timestamps = include_timestamps

priority property

priority: int

Return hook priority.

name property

name: str

Return hook name.

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Log invocation start.

Parameters:

Name	Type	Description	Default
prompt	str	User prompt	required
state	AgentState	Agent state	required

Returns:

Type	Description
AgentState	Unchanged state

Source code in src/locus/hooks/builtin/logging.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Log invocation start.

    Args:
        prompt: User prompt
        state: Agent state

    Returns:
        Unchanged state
    """
    prompt_preview = prompt[:100] + "..." if len(prompt) > 100 else prompt
    self._log(
        "Agent invocation starting",
        run_id=state.run_id,
        agent_id=state.agent_id,
        prompt_preview=prompt_preview,
        prompt_length=len(prompt),
    )
    return state

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

Log invocation completion.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution succeeded	required

Source code in src/locus/hooks/builtin/logging.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Log invocation completion.

    Args:
        state: Final agent state
        success: Whether execution succeeded
    """
    duration_ms = (state.updated_at - state.started_at).total_seconds() * 1000
    self._log(
        "Agent invocation completed",
        run_id=state.run_id,
        agent_id=state.agent_id,
        success=success,
        iterations=state.iteration,
        confidence=state.confidence,
        tool_calls=len(state.tool_executions),
        errors=len(state.errors),
        duration_ms=duration_ms,
    )

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent) -> None

Log tool call start.

Parameters:

Name	Type	Description	Default
event	BeforeToolCallEvent	Write-protected event carrying tool_name and arguments. The hook only inspects them.	required

Source code in src/locus/hooks/builtin/logging.py

async def on_before_tool_call(self, event: BeforeToolCallEvent) -> None:
    """Log tool call start.

    Args:
        event: Write-protected event carrying ``tool_name`` and
            ``arguments``. The hook only inspects them.
    """
    log_data: dict[str, Any] = {"tool_name": event.tool_name}
    if self._log_arguments:
        log_data["arguments"] = event.arguments
    else:
        log_data["argument_keys"] = list(event.arguments.keys())

    self._log("Tool call starting", **log_data)

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

Log tool call completion.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event carrying tool_name, result, and error. The hook only inspects them.	required

Source code in src/locus/hooks/builtin/logging.py

async def on_after_tool_call(self, event: AfterToolCallEvent) -> None:
    """Log tool call completion.

    Args:
        event: Write-protected event carrying ``tool_name``,
            ``result``, and ``error``. The hook only inspects them.
    """
    log_data: dict[str, Any] = {
        "tool_name": event.tool_name,
        "success": event.error is None,
    }

    if event.error:
        log_data["error"] = event.error
    elif self._log_results and event.result is not None:
        result_str = str(event.result)
        log_data["result_preview"] = (
            result_str[:200] + "..." if len(result_str) > 200 else result_str
        )
        log_data["result_length"] = len(result_str)

    self._log("Tool call completed", **log_data)

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Log iteration start.

Parameters:

Name	Type	Description	Default
iteration	int	Iteration number	required
state	AgentState	Current state	required

Source code in src/locus/hooks/builtin/logging.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Log iteration start.

    Args:
        iteration: Iteration number
        state: Current state
    """
    self._log(
        "Iteration starting",
        run_id=state.run_id,
        iteration=iteration,
        max_iterations=state.max_iterations,
        confidence=state.confidence,
    )

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Log iteration end.

Parameters:

Name	Type	Description	Default
iteration	int	Iteration number	required
state	AgentState	Current state	required

Source code in src/locus/hooks/builtin/logging.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Log iteration end.

    Args:
        iteration: Iteration number
        state: Current state
    """
    self._log(
        "Iteration completed",
        run_id=state.run_id,
        iteration=iteration,
        confidence=state.confidence,
        messages=len(state.messages),
    )

on_before_model_call async

on_before_model_call(event: BeforeModelCallEvent) -> None

Called before each model.complete() call.

Modify event.messages to change what the model sees. event.tools is read-only (inspect only).

Parameters:

Name	Type	Description	Default
event	BeforeModelCallEvent	Write-protected event. Writable: messages.	required

Source code in src/locus/hooks/provider.py

async def on_before_model_call(
    self,
    event: BeforeModelCallEvent,
) -> None:
    """Called before each model.complete() call.

    Modify event.messages to change what the model sees.
    event.tools is read-only (inspect only).

    Args:
        event: Write-protected event. Writable: messages.
    """

on_after_model_call async

on_after_model_call(event: AfterModelCallEvent) -> None

Called after each model.complete() call.

Set event.retry = True to discard response and re-call. Set event.response to replace the response. event.messages is read-only.

Parameters:

Name	Type	Description	Default
event	AfterModelCallEvent	Write-protected event. Writable: response, retry.	required

Source code in src/locus/hooks/provider.py

async def on_after_model_call(
    self,
    event: AfterModelCallEvent,
) -> None:
    """Called after each model.complete() call.

    Set event.retry = True to discard response and re-call.
    Set event.response to replace the response.
    event.messages is read-only.

    Args:
        event: Write-protected event. Writable: response, retry.
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in src/locus/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }

TelemetryHook

TelemetryHook(service_name: str = 'locus-agent', tracer_name: str = 'locus.hooks.telemetry', meter_name: str = 'locus.hooks.telemetry', record_arguments: bool = False, record_results: bool = False, priority: int = HookPriority.OBSERVABILITY_MIN + 10)

Bases: HookProvider

Hook provider for OpenTelemetry tracing and metrics.

Provides automatic instrumentation for: - Trace spans for agent invocations and iterations - Trace spans for tool calls - Metrics for invocation duration, tool call counts, etc.

Requires the telemetry extra: pip install locus[telemetry]

Example

from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import ConsoleSpanExporter, BatchSpanProcessor

Configure OpenTelemetry

provider = TracerProvider() provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter())) trace.set_tracer_provider(provider)

Add telemetry hook

registry.add_provider(TelemetryHook())

Initialize telemetry hook.

Parameters:

Name	Type	Description	Default
service_name	str	Service name for telemetry	'locus-agent'
tracer_name	str	Name for the OpenTelemetry tracer	'locus.hooks.telemetry'
meter_name	str	Name for the OpenTelemetry meter	'locus.hooks.telemetry'
record_arguments	bool	Whether to record tool arguments as span attributes	False
record_results	bool	Whether to record tool results as span attributes	False
priority	int	Hook priority (default: early in observability range)	OBSERVABILITY_MIN + 10

Raises:

Type	Description
ImportError	If OpenTelemetry is not installed

Source code in src/locus/hooks/builtin/telemetry.py

def __init__(
    self,
    service_name: str = "locus-agent",
    tracer_name: str = "locus.hooks.telemetry",
    meter_name: str = "locus.hooks.telemetry",
    record_arguments: bool = False,
    record_results: bool = False,
    priority: int = HookPriority.OBSERVABILITY_MIN + 10,
) -> None:
    """Initialize telemetry hook.

    Args:
        service_name: Service name for telemetry
        tracer_name: Name for the OpenTelemetry tracer
        meter_name: Name for the OpenTelemetry meter
        record_arguments: Whether to record tool arguments as span attributes
        record_results: Whether to record tool results as span attributes
        priority: Hook priority (default: early in observability range)

    Raises:
        ImportError: If OpenTelemetry is not installed
    """
    if not OTEL_AVAILABLE:
        msg = "OpenTelemetry is not installed. Install with: pip install locus[telemetry]"
        raise ImportError(msg)

    self._service_name = service_name
    self._tracer = trace.get_tracer(tracer_name)
    self._meter = metrics.get_meter(meter_name)
    self._record_arguments = record_arguments
    self._record_results = record_results
    self._priority = priority

    # Active spans tracking
    self._invocation_span: Span | None = None
    self._iteration_spans: dict[int, Span] = {}
    self._tool_spans: dict[str, tuple[Span, float]] = {}

    # Metrics
    self._invocation_counter = self._meter.create_counter(
        "locus.invocations",
        description="Number of agent invocations",
        unit="1",
    )
    self._invocation_duration = self._meter.create_histogram(
        "locus.invocation.duration",
        description="Duration of agent invocations",
        unit="ms",
    )
    self._iteration_counter = self._meter.create_counter(
        "locus.iterations",
        description="Number of agent iterations",
        unit="1",
    )
    self._tool_call_counter = self._meter.create_counter(
        "locus.tool_calls",
        description="Number of tool calls",
        unit="1",
    )
    self._tool_call_duration = self._meter.create_histogram(
        "locus.tool_call.duration",
        description="Duration of tool calls",
        unit="ms",
    )
    self._tool_error_counter = self._meter.create_counter(
        "locus.tool_errors",
        description="Number of tool call errors",
        unit="1",
    )

priority property

priority: int

Return hook priority.

name property

name: str

Return hook name.

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Start invocation span.

Parameters:

Name	Type	Description	Default
prompt	str	User prompt	required
state	AgentState	Agent state	required

Returns:

Type	Description
AgentState	Unchanged state

Source code in src/locus/hooks/builtin/telemetry.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Start invocation span.

    Args:
        prompt: User prompt
        state: Agent state

    Returns:
        Unchanged state
    """
    self._invocation_span = self._tracer.start_span(
        "agent.invocation",
        attributes={
            "locus.run_id": state.run_id,
            "locus.agent_id": state.agent_id or "",
            "locus.prompt_length": len(prompt),
            "locus.max_iterations": state.max_iterations,
            "service.name": self._service_name,
        },
    )
    self._invocation_counter.add(1, {"agent_id": state.agent_id or "default"})
    return state

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

End invocation span.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution succeeded	required

Source code in src/locus/hooks/builtin/telemetry.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """End invocation span.

    Args:
        state: Final agent state
        success: Whether execution succeeded
    """
    if self._invocation_span:
        duration_ms = (state.updated_at - state.started_at).total_seconds() * 1000

        self._invocation_span.set_attributes(
            {
                "locus.success": success,
                "locus.iterations": state.iteration,
                "locus.confidence": state.confidence,
                "locus.tool_calls": len(state.tool_executions),
                "locus.errors": len(state.errors),
                "locus.duration_ms": duration_ms,
            }
        )

        if success:
            self._invocation_span.set_status(Status(StatusCode.OK))
        else:
            self._invocation_span.set_status(
                Status(StatusCode.ERROR, "Agent invocation failed")
            )

        self._invocation_span.end()
        self._invocation_span = None

        # Record duration metric
        self._invocation_duration.record(
            duration_ms,
            {
                "agent_id": state.agent_id or "default",
                "success": str(success),
            },
        )

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent) -> None

Start tool call span.

Parameters:

Name	Type	Description	Default
event	BeforeToolCallEvent	Write-protected event carrying tool_name and arguments. The hook only inspects them.	required

Source code in src/locus/hooks/builtin/telemetry.py

async def on_before_tool_call(self, event: BeforeToolCallEvent) -> None:
    """Start tool call span.

    Args:
        event: Write-protected event carrying ``tool_name`` and
            ``arguments``. The hook only inspects them.
    """
    tool_name = event.tool_name
    span_attrs: dict[str, Any] = {
        "locus.tool_name": tool_name,
    }

    if self._record_arguments:
        # Sanitize arguments for span attributes
        for key, value in event.arguments.items():
            attr_key = f"locus.tool.arg.{key}"
            try:
                span_attrs[attr_key] = str(value)[:1000]  # Limit length
            except Exception:  # noqa: BLE001 — arbitrary user values; fall back to placeholder
                span_attrs[attr_key] = "<non-serializable>"

    span = self._tracer.start_span(f"tool.{tool_name}", attributes=span_attrs)
    self._tool_spans[tool_name] = (span, time.perf_counter())

    self._tool_call_counter.add(1, {"tool_name": tool_name})

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

End tool call span.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event carrying tool_name, result, and error.	required

Source code in src/locus/hooks/builtin/telemetry.py

async def on_after_tool_call(self, event: AfterToolCallEvent) -> None:
    """End tool call span.

    Args:
        event: Write-protected event carrying ``tool_name``,
            ``result``, and ``error``.
    """
    tool_name = event.tool_name
    error = event.error
    result = event.result
    if tool_name in self._tool_spans:
        span, start_time = self._tool_spans.pop(tool_name)
        duration_ms = (time.perf_counter() - start_time) * 1000

        span.set_attribute("locus.duration_ms", duration_ms)

        if error:
            span.set_status(Status(StatusCode.ERROR, error))
            span.set_attribute("locus.error", error[:1000])
            self._tool_error_counter.add(1, {"tool_name": tool_name})
        else:
            span.set_status(Status(StatusCode.OK))
            if self._record_results and result is not None:
                result_str = str(result)
                span.set_attribute("locus.result_preview", result_str[:500])

        span.end()

        self._tool_call_duration.record(
            duration_ms,
            {
                "tool_name": tool_name,
                "success": str(error is None),
            },
        )

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Start iteration span.

Parameters:

Name	Type	Description	Default
iteration	int	Iteration number	required
state	AgentState	Current state	required

Source code in src/locus/hooks/builtin/telemetry.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Start iteration span.

    Args:
        iteration: Iteration number
        state: Current state
    """
    span = self._tracer.start_span(
        f"agent.iteration.{iteration}",
        attributes={
            "locus.iteration": iteration,
            "locus.confidence": state.confidence,
            "locus.messages": len(state.messages),
        },
    )
    self._iteration_spans[iteration] = span
    self._iteration_counter.add(1, {"agent_id": state.agent_id or "default"})

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

End iteration span.

Parameters:

Name	Type	Description	Default
iteration	int	Iteration number	required
state	AgentState	Current state	required

Source code in src/locus/hooks/builtin/telemetry.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """End iteration span.

    Args:
        iteration: Iteration number
        state: Current state
    """
    if iteration in self._iteration_spans:
        span = self._iteration_spans.pop(iteration)
        span.set_attributes(
            {
                "locus.confidence_after": state.confidence,
                "locus.messages_after": len(state.messages),
            }
        )
        span.set_status(Status(StatusCode.OK))
        span.end()

on_before_model_call async

on_before_model_call(event: BeforeModelCallEvent) -> None

Called before each model.complete() call.

Modify event.messages to change what the model sees. event.tools is read-only (inspect only).

Parameters:

Name	Type	Description	Default
event	BeforeModelCallEvent	Write-protected event. Writable: messages.	required

Source code in src/locus/hooks/provider.py

async def on_before_model_call(
    self,
    event: BeforeModelCallEvent,
) -> None:
    """Called before each model.complete() call.

    Modify event.messages to change what the model sees.
    event.tools is read-only (inspect only).

    Args:
        event: Write-protected event. Writable: messages.
    """

on_after_model_call async

on_after_model_call(event: AfterModelCallEvent) -> None

Called after each model.complete() call.

Set event.retry = True to discard response and re-call. Set event.response to replace the response. event.messages is read-only.

Parameters:

Name	Type	Description	Default
event	AfterModelCallEvent	Write-protected event. Writable: response, retry.	required

Source code in src/locus/hooks/provider.py

async def on_after_model_call(
    self,
    event: AfterModelCallEvent,
) -> None:
    """Called after each model.complete() call.

    Set event.retry = True to discard response and re-call.
    Set event.response to replace the response.
    event.messages is read-only.

    Args:
        event: Write-protected event. Writable: response, retry.
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in src/locus/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }

NoOpTelemetryHook

NoOpTelemetryHook(priority: int = HookPriority.OBSERVABILITY_MIN + 10)

Bases: HookProvider

No-op telemetry hook for when OpenTelemetry is not available.

This hook does nothing but can be used as a drop-in replacement for TelemetryHook when telemetry is disabled.

Initialize no-op hook.

Parameters:

Name	Type	Description	Default
priority	int	Hook priority	OBSERVABILITY_MIN + 10

Source code in src/locus/hooks/builtin/telemetry.py

def __init__(self, priority: int = HookPriority.OBSERVABILITY_MIN + 10) -> None:
    """Initialize no-op hook.

    Args:
        priority: Hook priority
    """
    self._priority = priority

priority property

priority: int

Return hook priority.

name property

name: str

Return hook name.

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Called before agent starts processing.

Parameters:

Name	Type	Description	Default
prompt	str	The user prompt being processed	required
state	AgentState	Current agent state	required

Returns:

Type	Description
AgentState	Potentially modified agent state

Source code in src/locus/hooks/provider.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Called before agent starts processing.

    Args:
        prompt: The user prompt being processed
        state: Current agent state

    Returns:
        Potentially modified agent state
    """
    return state

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

Called after agent completes processing.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution completed successfully	required

Source code in src/locus/hooks/provider.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Called after agent completes processing.

    Args:
        state: Final agent state
        success: Whether execution completed successfully
    """

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent) -> None

Called before tool execution.

Modify event.arguments to change tool inputs. Set event.cancel = True or a string reason to skip execution. event.tool_name and event.tool_call_id are read-only.

Parameters:

Name	Type	Description	Default
event	BeforeToolCallEvent	Write-protected event. Writable: arguments, cancel.	required

Source code in src/locus/hooks/provider.py

async def on_before_tool_call(
    self,
    event: BeforeToolCallEvent,
) -> None:
    """Called before tool execution.

    Modify event.arguments to change tool inputs.
    Set event.cancel = True or a string reason to skip execution.
    event.tool_name and event.tool_call_id are read-only.

    Args:
        event: Write-protected event. Writable: arguments, cancel.
    """

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

Called after tool execution.

Set event.retry = True to re-execute the tool. Set event.result to replace the tool result. event.tool_name and event.error are read-only.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event. Writable: result, retry.	required

Source code in src/locus/hooks/provider.py

async def on_after_tool_call(
    self,
    event: AfterToolCallEvent,
) -> None:
    """Called after tool execution.

    Set event.retry = True to re-execute the tool.
    Set event.result to replace the tool result.
    event.tool_name and event.error are read-only.

    Args:
        event: Write-protected event. Writable: result, retry.
    """

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Called at the start of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in src/locus/hooks/provider.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the start of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Called at the end of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in src/locus/hooks/provider.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the end of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_before_model_call async

on_before_model_call(event: BeforeModelCallEvent) -> None

Called before each model.complete() call.

Modify event.messages to change what the model sees. event.tools is read-only (inspect only).

Parameters:

Name	Type	Description	Default
event	BeforeModelCallEvent	Write-protected event. Writable: messages.	required

Source code in src/locus/hooks/provider.py

async def on_before_model_call(
    self,
    event: BeforeModelCallEvent,
) -> None:
    """Called before each model.complete() call.

    Modify event.messages to change what the model sees.
    event.tools is read-only (inspect only).

    Args:
        event: Write-protected event. Writable: messages.
    """

on_after_model_call async

on_after_model_call(event: AfterModelCallEvent) -> None

Called after each model.complete() call.

Set event.retry = True to discard response and re-call. Set event.response to replace the response. event.messages is read-only.

Parameters:

Name	Type	Description	Default
event	AfterModelCallEvent	Write-protected event. Writable: response, retry.	required

Source code in src/locus/hooks/provider.py

async def on_after_model_call(
    self,
    event: AfterModelCallEvent,
) -> None:
    """Called after each model.complete() call.

    Set event.retry = True to discard response and re-call.
    Set event.response to replace the response.
    event.messages is read-only.

    Args:
        event: Write-protected event. Writable: response, retry.
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in src/locus/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }

ModelRetryHook

ModelRetryHook(max_retries: int = 3, initial_delay: float = 1.0, max_delay: float = 30.0, backoff_factor: float = 2.0, retry_on_empty: bool = True, priority: int = HookPriority.DEFAULT)

Bases: HookProvider

Retry model calls on throttle/rate limit with exponential backoff.

Catches empty responses and rate limit indicators, sets event.retry=True to trigger automatic re-invocation with increasing delays.

Works with all providers (OCI, OpenAI, Anthropic, Ollama).

Parameters:

Name	Type	Description	Default
max_retries	int	Maximum retry attempts per model call.	3
initial_delay	float	First retry delay in seconds.	1.0
max_delay	float	Maximum delay between retries.	30.0
backoff_factor	float	Multiplier for each subsequent delay.	2.0
retry_on_empty	bool	Retry when model returns empty content.	True

Source code in src/locus/hooks/builtin/retry.py

def __init__(
    self,
    max_retries: int = 3,
    initial_delay: float = 1.0,
    max_delay: float = 30.0,
    backoff_factor: float = 2.0,
    retry_on_empty: bool = True,
    priority: int = HookPriority.DEFAULT,
) -> None:
    self._max_retries = max_retries
    self._initial_delay = initial_delay
    self._max_delay = max_delay
    self._backoff_factor = backoff_factor
    self._retry_on_empty = retry_on_empty
    self._priority = priority
    self._attempt = 0
    self.retries_total = 0

on_before_model_call async

on_before_model_call(event: Any) -> None

Reset attempt counter before each new model call.

Source code in src/locus/hooks/builtin/retry.py

async def on_before_model_call(self, event: Any) -> None:
    """Reset attempt counter before each new model call."""
    self._attempt = 0

on_after_model_call async

on_after_model_call(event: Any) -> None

Check response and retry if needed.

Source code in src/locus/hooks/builtin/retry.py

async def on_after_model_call(self, event: Any) -> None:
    """Check response and retry if needed."""
    response = event.response
    content = response.message.content or ""
    has_tool_calls = bool(response.message.tool_calls)

    # Determine if we should retry
    should_retry = False

    if self._retry_on_empty and not content and not has_tool_calls:
        should_retry = True

    if not should_retry:
        # Successful response — reset
        self._attempt = 0
        return

    # Check retry budget
    if self._attempt >= self._max_retries:
        logger.warning(
            "ModelRetryHook: exhausted %d retries, accepting empty response",
            self._max_retries,
        )
        self._attempt = 0
        return

    # Calculate delay with exponential backoff
    delay = min(
        self._initial_delay * (self._backoff_factor**self._attempt),
        self._max_delay,
    )

    self._attempt += 1
    self.retries_total += 1

    logger.info(
        "ModelRetryHook: retry %d/%d after %.1fs delay (empty response)",
        self._attempt,
        self._max_retries,
        delay,
    )

    from locus.observability.emit import EV_HOOK_MODEL_RETRY, emit  # noqa: PLC0415

    await emit(
        EV_HOOK_MODEL_RETRY,
        attempt=self._attempt,
        max_retries=self._max_retries,
        delay_seconds=delay,
        reason="empty_response",
    )

    await asyncio.sleep(delay)
    event.retry = True

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Called before agent starts processing.

Parameters:

Name	Type	Description	Default
prompt	str	The user prompt being processed	required
state	AgentState	Current agent state	required

Returns:

Type	Description
AgentState	Potentially modified agent state

Source code in src/locus/hooks/provider.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Called before agent starts processing.

    Args:
        prompt: The user prompt being processed
        state: Current agent state

    Returns:
        Potentially modified agent state
    """
    return state

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

Called after agent completes processing.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution completed successfully	required

Source code in src/locus/hooks/provider.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Called after agent completes processing.

    Args:
        state: Final agent state
        success: Whether execution completed successfully
    """

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent) -> None

Called before tool execution.

Modify event.arguments to change tool inputs. Set event.cancel = True or a string reason to skip execution. event.tool_name and event.tool_call_id are read-only.

Parameters:

Name	Type	Description	Default
event	BeforeToolCallEvent	Write-protected event. Writable: arguments, cancel.	required

Source code in src/locus/hooks/provider.py

async def on_before_tool_call(
    self,
    event: BeforeToolCallEvent,
) -> None:
    """Called before tool execution.

    Modify event.arguments to change tool inputs.
    Set event.cancel = True or a string reason to skip execution.
    event.tool_name and event.tool_call_id are read-only.

    Args:
        event: Write-protected event. Writable: arguments, cancel.
    """

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

Called after tool execution.

Set event.retry = True to re-execute the tool. Set event.result to replace the tool result. event.tool_name and event.error are read-only.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event. Writable: result, retry.	required

Source code in src/locus/hooks/provider.py

async def on_after_tool_call(
    self,
    event: AfterToolCallEvent,
) -> None:
    """Called after tool execution.

    Set event.retry = True to re-execute the tool.
    Set event.result to replace the tool result.
    event.tool_name and event.error are read-only.

    Args:
        event: Write-protected event. Writable: result, retry.
    """

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Called at the start of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in src/locus/hooks/provider.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the start of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Called at the end of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in src/locus/hooks/provider.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the end of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in src/locus/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }

GuardrailsHook

GuardrailsHook(config: GuardrailConfig | None = None, on_violation: Callable[[GuardrailViolation], None] | None = None, priority: int = HookPriority.SECURITY_DEFAULT)

Bases: HookProvider

Hook provider for security guardrails.

Provides: - Input validation and filtering - Output sanitization - PII detection and redaction - Dangerous content blocking - Tool allowlist/blocklist enforcement

Example

config = GuardrailConfig( block_dangerous_tools=frozenset({"shell", "exec"}), default_action=GuardrailAction.BLOCK, ) registry.add_provider(GuardrailsHook(config))

Initialize guardrails hook.

Parameters:

Name	Type	Description	Default
config	GuardrailConfig | None	Guardrail configuration	None
on_violation	Callable[[GuardrailViolation], None] | None	Callback for violations (receives GuardrailViolation)	None
priority	int	Hook priority (default: middle of security range)	SECURITY_DEFAULT

Source code in src/locus/hooks/builtin/guardrails.py

def __init__(
    self,
    config: GuardrailConfig | None = None,
    on_violation: Callable[[GuardrailViolation], None] | None = None,
    priority: int = HookPriority.SECURITY_DEFAULT,
) -> None:
    """Initialize guardrails hook.

    Args:
        config: Guardrail configuration
        on_violation: Callback for violations (receives GuardrailViolation)
        priority: Hook priority (default: middle of security range)
    """
    self._config = config or GuardrailConfig()
    self._on_violation = on_violation
    self._priority = priority
    self._violations: list[GuardrailViolation] = []

    # Compile patterns for efficiency
    self._compiled_pii: dict[str, re.Pattern[str]] = {
        name: re.compile(pattern) for name, pattern in self._config.pii_patterns.items()
    }
    self._compiled_blocked: dict[str, re.Pattern[str]] = {
        name: re.compile(pattern)
        for name, pattern in self._config.blocked_content_patterns.items()
    }

priority property

priority: int

Return hook priority.

name property

name: str

Return hook name.

violations property

violations: list[GuardrailViolation]

Get recorded violations.

clear_violations

clear_violations() -> None

Clear recorded violations.

Source code in src/locus/hooks/builtin/guardrails.py

def clear_violations(self) -> None:
    """Clear recorded violations."""
    self._violations.clear()

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Validate input prompt.

Parameters:

Name	Type	Description	Default
prompt	str	User prompt	required
state	AgentState	Agent state	required

Returns:

Type	Description
AgentState	State, potentially with metadata about violations

Raises:

Type	Description
ValueError	If prompt is blocked

Source code in src/locus/hooks/builtin/guardrails.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Validate input prompt.

    Args:
        prompt: User prompt
        state: Agent state

    Returns:
        State, potentially with metadata about violations

    Raises:
        ValueError: If prompt is blocked
    """
    violations: list[GuardrailViolation] = []

    # Check prompt length
    if len(prompt) > self._config.max_prompt_length:
        violation = GuardrailViolation(
            rule_name="max_prompt_length",
            description=f"Prompt exceeds maximum length ({len(prompt)} > {self._config.max_prompt_length})",
            action=self._get_action("max_prompt_length"),
            location="input",
        )
        violations.append(violation)
        self._record_violation(violation)

    # Check for blocked content
    violations.extend(self._check_blocked_content(prompt, "input"))

    # Check for PII
    pii_violations = self._check_pii(prompt, "input")
    violations.extend(pii_violations)

    # Handle blocking
    if self._should_block(violations):
        msg = f"Input blocked by guardrails: {violations[0].description}"
        raise ValueError(msg)

    # Store violations in metadata
    if violations:
        state = state.with_metadata(
            "guardrail_violations",
            [
                {
                    "rule_name": v.rule_name,
                    "description": v.description,
                    "action": v.action.value,
                    "location": v.location,
                }
                for v in violations
            ],
        )

    return state

on_before_tool_call async

on_before_tool_call(event: BeforeToolCallEvent) -> None

Validate tool call.

Parameters:

Name	Type	Description	Default
event	BeforeToolCallEvent	Write-protected event. The hook may mutate event.arguments (PII redaction) or set event.cancel to short-circuit a blocked tool.	required

Raises:

Type	Description
ValueError	If tool is blocked

Source code in src/locus/hooks/builtin/guardrails.py

async def on_before_tool_call(self, event: BeforeToolCallEvent) -> None:
    """Validate tool call.

    Args:
        event: Write-protected event. The hook may mutate
            ``event.arguments`` (PII redaction) or set
            ``event.cancel`` to short-circuit a blocked tool.

    Raises:
        ValueError: If tool is blocked
    """
    tool_name = event.tool_name
    arguments = event.arguments

    # Check tool blocklist
    if tool_name in self._config.block_dangerous_tools:
        violation = GuardrailViolation(
            rule_name="blocked_tool",
            description=f"Tool '{tool_name}' is blocked",
            action=GuardrailAction.BLOCK,
            location="tool_args",
        )
        self._record_violation(violation)
        msg = f"Tool '{tool_name}' is blocked by guardrails"
        raise ValueError(msg)

    # Check tool allowlist
    if (
        self._config.allow_only_tools is not None
        and tool_name not in self._config.allow_only_tools
    ):
        violation = GuardrailViolation(
            rule_name="tool_not_allowed",
            description=f"Tool '{tool_name}' is not in allowlist",
            action=GuardrailAction.BLOCK,
            location="tool_args",
        )
        self._record_violation(violation)
        msg = f"Tool '{tool_name}' is not allowed"
        raise ValueError(msg)

    # Check arguments for dangerous content
    args_str = str(arguments)
    violations = self._check_blocked_content(args_str, "tool_args")

    if self._should_block(violations):
        msg = f"Tool arguments blocked: {violations[0].description}"
        raise ValueError(msg)

    # Check for and optionally redact PII in arguments
    pii_violations = self._check_pii(args_str, "tool_args")
    if pii_violations and any(v.action == GuardrailAction.REDACT for v in pii_violations):
        # Redact PII from string arguments — write back to the event
        # so downstream hooks and the executor see the redacted form.
        redacted_args: dict[str, Any] = {}
        for key, value in arguments.items():
            if isinstance(value, str):
                redacted_args[key] = self._redact_pii(value)
            else:
                redacted_args[key] = value
        event.arguments = redacted_args

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

Validate tool result.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event carrying tool_name, result, and error.	required

Source code in src/locus/hooks/builtin/guardrails.py

async def on_after_tool_call(self, event: AfterToolCallEvent) -> None:
    """Validate tool result.

    Args:
        event: Write-protected event carrying ``tool_name``,
            ``result``, and ``error``.
    """
    result = event.result
    if result is None:
        return

    result_str = str(result)

    # Check result length
    if len(result_str) > self._config.max_tool_result_length:
        violation = GuardrailViolation(
            rule_name="max_tool_result_length",
            description=(
                f"Tool result exceeds maximum length "
                f"({len(result_str)} > {self._config.max_tool_result_length})"
            ),
            action=self._get_action("max_tool_result_length"),
            location="tool_result",
        )
        self._record_violation(violation)

    # Check for PII in results
    self._check_pii(result_str, "tool_result")

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

Called after agent completes processing.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution completed successfully	required

Source code in src/locus/hooks/provider.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Called after agent completes processing.

    Args:
        state: Final agent state
        success: Whether execution completed successfully
    """

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Called at the start of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in src/locus/hooks/provider.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the start of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Called at the end of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in src/locus/hooks/provider.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the end of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_before_model_call async

on_before_model_call(event: BeforeModelCallEvent) -> None

Called before each model.complete() call.

Modify event.messages to change what the model sees. event.tools is read-only (inspect only).

Parameters:

Name	Type	Description	Default
event	BeforeModelCallEvent	Write-protected event. Writable: messages.	required

Source code in src/locus/hooks/provider.py

async def on_before_model_call(
    self,
    event: BeforeModelCallEvent,
) -> None:
    """Called before each model.complete() call.

    Modify event.messages to change what the model sees.
    event.tools is read-only (inspect only).

    Args:
        event: Write-protected event. Writable: messages.
    """

on_after_model_call async

on_after_model_call(event: AfterModelCallEvent) -> None

Called after each model.complete() call.

Set event.retry = True to discard response and re-call. Set event.response to replace the response. event.messages is read-only.

Parameters:

Name	Type	Description	Default
event	AfterModelCallEvent	Write-protected event. Writable: response, retry.	required

Source code in src/locus/hooks/provider.py

async def on_after_model_call(
    self,
    event: AfterModelCallEvent,
) -> None:
    """Called after each model.complete() call.

    Set event.retry = True to discard response and re-call.
    Set event.response to replace the response.
    event.messages is read-only.

    Args:
        event: Write-protected event. Writable: response, retry.
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in src/locus/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }

SteeringHook

SteeringHook(model: Any, policy: str = '', evaluate_tools: bool = True, evaluate_responses: bool = False, interrupt_tools: set[str] | None = None, priority: int = HookPriority.SECURITY_DEFAULT)

Bases: HookProvider

LLM-powered steering for real-time agent guidance.

Evaluates each tool call before execution using a separate LLM. The steering model decides whether to proceed, guide (cancel with feedback), or interrupt (pause for human).

Parameters:

Name	Type	Description	Default
model	Any	LLM for steering decisions (can be smaller/cheaper than main model).	required
policy	str	Natural language policy the agent must follow.	''
evaluate_tools	bool	If True, evaluate tool calls before execution.	True
evaluate_responses	bool	If True, evaluate model responses after generation.	False
interrupt_tools	set[str] | None	Tools that always require human approval.	None

Source code in src/locus/hooks/builtin/steering.py

def __init__(
    self,
    model: Any,
    policy: str = "",
    evaluate_tools: bool = True,
    evaluate_responses: bool = False,
    interrupt_tools: set[str] | None = None,
    priority: int = HookPriority.SECURITY_DEFAULT,
) -> None:
    self._model = model
    self._policy = policy
    self._evaluate_tools = evaluate_tools
    self._evaluate_responses = evaluate_responses
    self._interrupt_tools = interrupt_tools or set()
    self._priority = priority
    self._context = SteeringContext(policy=policy)
    self.decisions: list[SteeringDecision] = []

on_before_tool_call async

on_before_tool_call(event: Any) -> None

Evaluate tool call before execution.

Source code in src/locus/hooks/builtin/steering.py

async def on_before_tool_call(self, event: Any) -> None:
    """Evaluate tool call before execution."""
    if not self._evaluate_tools:
        return

    # Always interrupt for specified tools
    if event.tool_name in self._interrupt_tools:
        decision = SteeringDecision(
            action=SteeringAction.INTERRUPT,
            reason=f"Tool '{event.tool_name}' requires human approval",
        )
        self.decisions.append(decision)
        event.cancel = f"REQUIRES APPROVAL: {event.tool_name} needs human approval"
        return

    # LLM evaluation
    decision = await self._evaluate_tool_call(event.tool_name, event.arguments)
    self.decisions.append(decision)

    from locus.observability.emit import EV_HOOK_STEERING_APPLIED, emit  # noqa: PLC0415

    if decision.action == SteeringAction.GUIDE:
        event.cancel = decision.guidance
        logger.info(
            "Steering GUIDE: %s(%s) — %s", event.tool_name, event.arguments, decision.reason
        )
        await emit(
            EV_HOOK_STEERING_APPLIED,
            action="guide",
            tool_name=event.tool_name,
            reason=decision.reason,
        )

    elif decision.action == SteeringAction.INTERRUPT:
        event.cancel = f"REQUIRES APPROVAL: {decision.reason}"
        logger.info("Steering INTERRUPT: %s — %s", event.tool_name, decision.reason)
        await emit(
            EV_HOOK_STEERING_APPLIED,
            action="interrupt",
            tool_name=event.tool_name,
            reason=decision.reason,
        )

    # Record in context
    self._context.record_tool_call(event.tool_name, event.arguments)

on_before_model_call async

on_before_model_call(event: Any) -> None

Track model calls in context.

Source code in src/locus/hooks/builtin/steering.py

async def on_before_model_call(self, event: Any) -> None:
    """Track model calls in context."""
    self._context.model_calls += 1

on_after_model_call async

on_after_model_call(event: Any) -> None

Optionally evaluate model responses.

Source code in src/locus/hooks/builtin/steering.py

async def on_after_model_call(self, event: Any) -> None:
    """Optionally evaluate model responses."""
    if not self._evaluate_responses:
        return

    content = event.response.message.content or ""
    if not content:
        return

    # Simple policy check — could be expanded to use steering LLM
    if self._policy and any(
        word in content.lower() for word in ["password", "secret", "credential"]
    ):
        logger.warning("Steering: response may contain sensitive info")

on_before_invocation async

on_before_invocation(prompt: str, state: AgentState) -> AgentState

Called before agent starts processing.

Parameters:

Name	Type	Description	Default
prompt	str	The user prompt being processed	required
state	AgentState	Current agent state	required

Returns:

Type	Description
AgentState	Potentially modified agent state

Source code in src/locus/hooks/provider.py

async def on_before_invocation(
    self,
    prompt: str,
    state: AgentState,
) -> AgentState:
    """Called before agent starts processing.

    Args:
        prompt: The user prompt being processed
        state: Current agent state

    Returns:
        Potentially modified agent state
    """
    return state

on_after_invocation async

on_after_invocation(state: AgentState, success: bool) -> None

Called after agent completes processing.

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state	required
success	bool	Whether execution completed successfully	required

Source code in src/locus/hooks/provider.py

async def on_after_invocation(
    self,
    state: AgentState,
    success: bool,
) -> None:
    """Called after agent completes processing.

    Args:
        state: Final agent state
        success: Whether execution completed successfully
    """

on_after_tool_call async

on_after_tool_call(event: AfterToolCallEvent) -> None

Called after tool execution.

Set event.retry = True to re-execute the tool. Set event.result to replace the tool result. event.tool_name and event.error are read-only.

Parameters:

Name	Type	Description	Default
event	AfterToolCallEvent	Write-protected event. Writable: result, retry.	required

Source code in src/locus/hooks/provider.py

async def on_after_tool_call(
    self,
    event: AfterToolCallEvent,
) -> None:
    """Called after tool execution.

    Set event.retry = True to re-execute the tool.
    Set event.result to replace the tool result.
    event.tool_name and event.error are read-only.

    Args:
        event: Write-protected event. Writable: result, retry.
    """

on_iteration_start async

on_iteration_start(iteration: int, state: AgentState) -> None

Called at the start of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in src/locus/hooks/provider.py

async def on_iteration_start(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the start of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

on_iteration_end async

on_iteration_end(iteration: int, state: AgentState) -> None

Called at the end of each agent iteration.

Parameters:

Name	Type	Description	Default
iteration	int	Current iteration number (0-indexed)	required
state	AgentState	Current agent state	required

Source code in src/locus/hooks/provider.py

async def on_iteration_end(
    self,
    iteration: int,
    state: AgentState,
) -> None:
    """Called at the end of each agent iteration.

    Args:
        iteration: Current iteration number (0-indexed)
        state: Current agent state
    """

register_hooks

register_hooks() -> dict[str, bool]

Return which hooks this provider implements.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping hook names to whether they are implemented.
dict[str, bool]	Useful for optimization - registry can skip calling unimplemented hooks.

Source code in src/locus/hooks/provider.py

def register_hooks(self) -> dict[str, bool]:
    """Return which hooks this provider implements.

    Returns:
        Dictionary mapping hook names to whether they are implemented.
        Useful for optimization - registry can skip calling unimplemented hooks.
    """
    return {
        "on_before_invocation": True,
        "on_after_invocation": True,
        "on_before_tool_call": True,
        "on_after_tool_call": True,
        "on_iteration_start": True,
        "on_iteration_end": True,
        "on_before_model_call": True,
        "on_after_model_call": True,
    }