orchestrata_ultimate.core.adapters.claude.hooks

Reference › orchestrata_ultimate › core › adapters › claude › hooks

Hook management for the Claude Code integration. HookOrchestrator keeps a registry of hooks by HookCategory (PRE_TOOL, POST_TOOL, SESSION_BEGIN, SESSION_FINISH, SYSTEM_STOP). Each hook is a HookDescriptor with an identifier, a handler callable, an ExecutionPriority (URGENT 0, HIGH 10, MEDIUM 50, LOW 90), an active flag and an optional tool-name filter. register, unregister, activate_hook, deactivate_hook and list_hooks manage the registry.

Hooks can come from a JSON configuration file given to the constructor. _load_configuration reads its hooks list, and _resolve_handler turns a module.path:callable string into a callable. A hook with no handler, or one that cannot be resolved, is logged as an error and skipped rather than turned into a silent no-op. run_pre_tool_hooks walks the active pre-tool hooks whose tool filter matches and returns an InterceptorResult from tool_interceptor; the first hook that blocks ends the walk, and if none blocks the result is an ALLOW marked "Allowed by default". run_post_tool_hooks, run_session_begin_hooks and run_session_finish_hooks run the other categories. Every execution is recorded as an ExecutionOutcome (succeeded, verdict, note, elapsed time), available through get_execution_history. An optional ToolInterceptor can be supplied to the orchestrator.

At a glance

Class diagram

Dependency diagram

API

Classes

HookCategory (enum)

Bases: StrEnum · Line 27

Hook execution categories.

ExecutionPriority (enum)

Bases: IntEnum · Line 37

Hook execution priority ordering.

HookDescriptor (dataclass)

Decorators: dataclass · Line 47

Describes a registered hook.

ExecutionOutcome (dataclass)

Decorators: dataclass · Line 60

Result from hook execution.

HookOrchestrator

Line 71

Orchestrates hook execution for Claude Code.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/adapters/claude/hooks.py

# mypy: ignore-errors
"""Hook management for Claude Code integration.

Completely rewritten with fresh architecture.
"""

import importlib
import json
import logging
import re
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import IntEnum, StrEnum
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.adapters.claude.tool_interceptor import (
    InterceptorDecision,
    InterceptorResult,
    ToolInterceptor,
)

log = logging.getLogger(__name__)


class HookCategory(StrEnum):
    """Hook execution categories."""

    PRE_TOOL = "pre_tool"
    POST_TOOL = "post_tool"
    SESSION_BEGIN = "session_begin"
    SESSION_FINISH = "session_finish"
    SYSTEM_STOP = "system_stop"


class ExecutionPriority(IntEnum):
    """Hook execution priority ordering."""

    URGENT = 0
    HIGH = 10
    MEDIUM = 50
    LOW = 90


@dataclass
class HookDescriptor:
    """Describes a registered hook."""

    identifier: str
    category: HookCategory
    handler: Callable
    priority: ExecutionPriority = ExecutionPriority.MEDIUM
    active: bool = True
    tool_filter: str | None = None
    metadata: dict = field(default_factory=dict)


@dataclass
class ExecutionOutcome:
    """Result from hook execution."""

    hook_id: str
    succeeded: bool
    verdict: InterceptorDecision = InterceptorDecision.ALLOW
    note: str = ""
    elapsed_ms: float = 0.0
    extra: dict = field(default_factory=dict)


class HookOrchestrator:
    """Orchestrates hook execution for Claude Code.

    Manages hook lifecycle, execution ordering, and governance integration.
    """

    def __init__(
        self,
        config_file: Path | None = None,
        interceptor: ToolInterceptor | None = None,
    ) -> None:
        """Initialize orchestrator.

        Args:
            config_file: Optional configuration file path.
            interceptor: Optional tool interceptor instance.
        """
        self.config_file = config_file
        self.interceptor = interceptor or ToolInterceptor()
        self._registry: dict[HookCategory, list[HookDescriptor]] = {
            cat: [] for cat in HookCategory
        }
        self._execution_history: list[ExecutionOutcome] = []

        if config_file and config_file.exists():
            self._load_configuration(config_file)

    def _load_configuration(self, path: Path) -> None:
        """Load hook configuration from file.

        Args:
            path: Configuration file path.
        """
        try:
            raw = path.read_text(encoding="utf-8")
            config = json.loads(raw)
            log.info(f"Loaded hook config from {path}")

            for hook_def in config.get("hooks", []):
                category = HookCategory(hook_def["category"])
                identifier = hook_def.get("identifier", "<unknown>")
                spec = hook_def.get("handler")
                if not spec:
                    # No silent no-op: a configured hook with no handler is a
                    # config error — surface it loudly and skip it.
                    log.error(
                        "Hook %r declares no 'handler' (module:callable); "
                        "skipping — configured hooks must resolve to a real "
                        "callable.",
                        identifier,
                    )
                    continue
                try:
                    handler = self._resolve_handler(spec)
                except Exception as exc:
                    log.error(
                        "Hook %r handler %r could not be resolved: %s; skipping.",
                        identifier,
                        spec,
                        exc,
                    )
                    continue
                self._registry[category].append(
                    HookDescriptor(
                        identifier=hook_def["identifier"],
                        category=category,
                        handler=handler,
                        priority=ExecutionPriority[hook_def.get("priority", "MEDIUM")],
                        active=hook_def.get("active", True),
                        tool_filter=hook_def.get("tool_filter"),
                        metadata=hook_def.get("metadata", {}),
                    )
                )
        except Exception as e:
            log.error(f"Failed loading config: {e}", exc_info=True)

    @staticmethod
    def _resolve_handler(spec: str) -> Callable:
        """Resolve a ``'module.path:callable'`` string to a real callable.

        Args:
            spec: Dotted import path of the form ``package.module:function``.

        Returns:
            The resolved callable.

        Raises:
            ValueError: If *spec* is not ``module:callable`` form or the target
                is not callable.
            ImportError / AttributeError: If the module/attribute cannot be
                imported.
        """
        if ":" not in spec:
            raise ValueError(f"hook handler must be 'module:callable', got {spec!r}")
        module_path, _, attr = spec.partition(":")
        module = importlib.import_module(module_path)
        handler = getattr(module, attr)
        if not callable(handler):
            raise ValueError(f"hook handler {spec!r} is not callable")
        return handler

    def register(
        self,
        identifier: str,
        category: HookCategory,
        handler: Callable,
        priority: ExecutionPriority = ExecutionPriority.MEDIUM,
        tool_filter: str | None = None,
    ) -> None:
        """Register new hook.

        Args:
            identifier: Unique hook identifier.
            category: Hook category.
            handler: Callback function.
            priority: Execution priority.
            tool_filter: Optional regex for tool filtering.
        """
        descriptor = HookDescriptor(
            identifier=identifier,
            category=category,
            handler=handler,
            priority=priority,
            tool_filter=tool_filter,
        )
        self._registry[category].append(descriptor)
        self._registry[category].sort(key=lambda h: h.priority)
        log.debug(f"Registered hook: {identifier} ({category.value})")

    def unregister(self, identifier: str, category: HookCategory) -> bool:
        """Remove hook by identifier.

        Args:
            identifier: Hook identifier.
            category: Hook category.

        Returns:
            True if hook was removed.
        """
        hooks = self._registry[category]
        for idx, hook in enumerate(hooks):
            if hook.identifier == identifier:
                hooks.pop(idx)
                log.debug(f"Unregistered hook: {identifier}")
                return True
        return False

    def list_hooks(self, category: HookCategory) -> list[HookDescriptor]:
        """Get hooks for category.

        Args:
            category: Hook category.

        Returns:
            List of hook descriptors.
        """
        return self._registry[category].copy()

    def activate_hook(self, identifier: str, category: HookCategory) -> bool:
        """Activate hook.

        Args:
            identifier: Hook identifier.
            category: Hook category.

        Returns:
            True if found and activated.
        """
        for hook in self._registry[category]:
            if hook.identifier == identifier:
                hook.active = True
                return True
        return False

    def deactivate_hook(self, identifier: str, category: HookCategory) -> bool:
        """Deactivate hook.

        Args:
            identifier: Hook identifier.
            category: Hook category.

        Returns:
            True if found and deactivated.
        """
        for hook in self._registry[category]:
            if hook.identifier == identifier:
                hook.active = False
                return True
        return False

    async def run_pre_tool_hooks(
        self,
        tool: str,
        parameters: dict[str, Any],
    ) -> InterceptorResult:
        """Execute pre-tool hooks.

        Args:
            tool: Tool name.
            parameters: Tool parameters.

        Returns:
            Interceptor result.
        """
        start = datetime.now(UTC)

        # Check interceptor first
        result = await self.interceptor.intercept(tool, parameters)

        if result.decision == InterceptorDecision.BLOCK:
            self._record_outcome(
                "interceptor",
                False,
                result.decision,
                result.message,
                start,
            )
            return result

        return await self._run_pre_tool_hook_loop(tool, parameters, start)

    async def _run_pre_tool_hook_loop(
        self,
        tool: str,
        parameters: dict[str, Any],
        start: datetime,
    ) -> InterceptorResult:
        """Iterate over registered pre-tool hooks and apply each one.

        Args:
            tool: Tool name.
            parameters: Tool parameters.
            start: Start timestamp for elapsed tracking.

        Returns:
            Interceptor result (BLOCK or default ALLOW).
        """
        for hook in self._registry[HookCategory.PRE_TOOL]:
            if not hook.active:
                continue

            if hook.tool_filter and not self._matches_filter(tool, hook.tool_filter):
                continue

            block_result = await self._apply_pre_tool_hook(
                hook, tool, parameters, start
            )
            if block_result is not None:
                return block_result

        return InterceptorResult(
            decision=InterceptorDecision.ALLOW,
            reason="Allowed by default",
            tool_name=tool,
            original_input=parameters,
        )

    async def _apply_pre_tool_hook(
        self,
        hook: HookDescriptor,
        tool: str,
        parameters: dict[str, Any],
        start: datetime,
    ) -> InterceptorResult | None:
        """Execute a single pre-tool hook and return a block result if blocked.

        Args:
            hook: Hook descriptor to execute.
            tool: Tool name.
            parameters: Tool parameters.
            start: Start timestamp for elapsed tracking.

        Returns:
            InterceptorResult if the hook blocks, None otherwise.
        """
        try:
            outcome = await self._execute_hook(hook, tool, parameters)
            self._record_outcome(
                hook.identifier,
                outcome.succeeded,
                outcome.verdict,
                outcome.note,
                start,
            )

            if outcome.verdict == InterceptorDecision.BLOCK:
                return InterceptorResult(
                    decision=InterceptorDecision.BLOCK,
                    reason=outcome.note or "Blocked by hook",
                    tool_name=tool,
                    original_input=parameters,
                    message=outcome.note or "",
                    hook_name=hook.identifier,
                )
        except Exception as e:  # pragma: no cover
            log.error(
                f"Hook {hook.identifier} failed: {e}", exc_info=True
            )  # pragma: no cover
            self._record_outcome(
                hook.identifier, False, InterceptorDecision.ALLOW, str(e), start
            )  # pragma: no cover

        return None

    async def run_post_tool_hooks(
        self,
        tool: str,
        parameters: dict[str, Any],
        output: Any,
    ) -> None:
        """Execute post-tool hooks.

        Args:
            tool: Tool name.
            parameters: Tool parameters.
            output: Tool output.
        """
        for hook in self._registry[HookCategory.POST_TOOL]:
            if not hook.active:
                continue

            if hook.tool_filter and not self._matches_filter(tool, hook.tool_filter):
                continue

            try:
                await self._execute_hook(hook, tool, parameters, output)
            except Exception as e:  # pragma: no cover
                log.error(
                    f"Post-tool hook {hook.identifier} failed: {e}", exc_info=True
                )  # pragma: no cover

    async def run_session_begin_hooks(self, session_id: str) -> None:
        """Execute session begin hooks.

        Args:
            session_id: Session identifier.
        """
        for hook in self._registry[HookCategory.SESSION_BEGIN]:
            if not hook.active:
                continue

            try:
                await self._execute_hook(hook, session_id=session_id)
            except Exception as e:  # pragma: no cover
                log.error(
                    f"Session begin hook {hook.identifier} failed: {e}", exc_info=True
                )  # pragma: no cover

    async def run_session_finish_hooks(self, session_id: str) -> None:
        """Execute session finish hooks.

        Args:
            session_id: Session identifier.
        """
        for hook in self._registry[HookCategory.SESSION_FINISH]:
            if not hook.active:
                continue

            try:
                await self._execute_hook(hook, session_id=session_id)
            except Exception as e:  # pragma: no cover
                log.error(
                    f"Session finish hook {hook.identifier} failed: {e}", exc_info=True
                )  # pragma: no cover

    async def _execute_hook(
        self,
        hook: HookDescriptor,
        *args,
        **kwargs,
    ) -> ExecutionOutcome:
        """Execute single hook.

        Args:
            hook: Hook descriptor.
            *args: Positional arguments.
            **kwargs: Keyword arguments.

        Returns:
            Execution outcome.
        """
        start = datetime.now(UTC)
        try:
            outcome = hook.handler(*args, **kwargs)
            if hasattr(outcome, "__await__"):
                outcome = await outcome

            elapsed = (datetime.now(UTC) - start).total_seconds() * 1000

            if isinstance(outcome, ExecutionOutcome):
                outcome.elapsed_ms = elapsed
                return outcome

            return ExecutionOutcome(
                hook_id=hook.identifier,
                succeeded=True,
                verdict=InterceptorDecision.ALLOW,
                elapsed_ms=elapsed,
            )
        except Exception as e:
            elapsed = (datetime.now(UTC) - start).total_seconds() * 1000
            return ExecutionOutcome(
                hook_id=hook.identifier,
                succeeded=False,
                verdict=InterceptorDecision.ALLOW,
                note=str(e),
                elapsed_ms=elapsed,
            )

    def _matches_filter(self, tool: str, pattern: str) -> bool:
        """Check if tool matches filter pattern.

        Args:
            tool: Tool name.
            pattern: Regex pattern.

        Returns:
            True if matches.
        """
        return bool(re.match(pattern, tool))

    def _record_outcome(
        self,
        hook_id: str,
        succeeded: bool,
        verdict: InterceptorDecision,
        note: str,
        start: datetime,
    ) -> None:
        """Record hook execution outcome.

        Args:
            hook_id: Hook identifier.
            succeeded: Success status.
            verdict: Interceptor verdict.
            note: Outcome message.
            start: Start timestamp.
        """
        elapsed = (datetime.now(UTC) - start).total_seconds() * 1000
        outcome = ExecutionOutcome(
            hook_id=hook_id,
            succeeded=succeeded,
            verdict=verdict,
            note=note,
            elapsed_ms=elapsed,
        )
        self._execution_history.append(outcome)
        if len(self._execution_history) > 1000:
            self._execution_history = self._execution_history[-500:]

    def get_execution_history(self) -> list[ExecutionOutcome]:
        """Get execution history.

        Returns:
            List of outcomes.
        """
        return self._execution_history.copy()

    def clear_history(self) -> None:
        """Clear execution history."""
        self._execution_history.clear()


# Backwards compatibility alias
HookManager = HookOrchestrator