orchestrata_ultimate.session.hook_registry

Reference › orchestrata_ultimate › session › hook_registry

A registry of session lifecycle hooks (spec 006). SessionHook is the protocol every callback follows. HookRegistry maps a named hook point to a list of callbacks: add registers one, remove removes it and returns whether it was found, fire(hook_point, session_id, **kwargs) calls every callback registered at that point, count reports how many there are, and clear removes all of them or those of one point. These are in-process Python callbacks, unrelated to the Claude Code hooks in the hooks package.

At a glance

Class diagram

Dependency diagram

API

Classes

SessionHook (protocol)

Bases: Protocol · Decorators: traces(spec='006', story='US-006', reqs=['REQ-006-FR11-002', 'REQ-006-US-006-003', 'REQ-006-US-006-006']), runtime_checkable · Line 38

Protocol defining the required hook callback signature.

HookRegistry

Line 56

Registry for session lifecycle hooks.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/hook_registry.py

# mypy: ignore-errors
"""Session hook registry for spec 006.

Provides a typed hook system where every hook callback receives
``(session_id: str, **kwargs)`` — enforcing a uniform signature
across all session lifecycle events.

Hook points:
    - ``on_session_open``  — fired after a session is created
    - ``on_session_close`` — fired after a session is closed
    - ``on_fact_emitted``  — fired after a fact is emitted

Example:
    >>> registry = HookRegistry()
    >>> def my_hook(session_id: str, **kwargs) -> None:
    ...     print(f"Session {session_id} event: {kwargs}")
    >>> registry.add("on_session_open", my_hook)
    >>> registry.fire("on_session_open", "session_20260223_141500_a3f8b21c", purpose="demo")
"""

import logging
from typing import Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)

# Valid hook point names
HOOK_POINTS = frozenset({"on_session_open", "on_session_close", "on_fact_emitted"})


@traces(
    spec="006",
    story="US-006",
    reqs=["REQ-006-FR11-002", "REQ-006-US-006-003", "REQ-006-US-006-006"],
)
@runtime_checkable
class SessionHook(Protocol):
    """Protocol defining the required hook callback signature.

    Every session hook must accept ``session_id`` as the first positional
    argument and arbitrary keyword arguments for event-specific data.

    Args:
        session_id: The session identifier (spec 006 format).
        **kwargs: Event-specific keyword arguments.

    Returns:
        None.
    """

    @traces(spec="006", story="US-006", reqs=["REQ-006-FR11-002", "REQ-006-US-006-006"])
    def __call__(self, session_id: str, **kwargs) -> None: ...


class HookRegistry:
    """Registry for session lifecycle hooks.

    Manages registration, removal, and firing of hooks at defined
    hook points. All hooks share the signature
    ``hook_function(session_id: str, **kwargs) -> None``.

    Attributes:
        _hooks: Mapping from hook point name to list of registered callbacks.

    Example:
        >>> reg = HookRegistry()
        >>> calls = []
        >>> reg.add("on_session_open", lambda session_id, **kw: calls.append(session_id))
        >>> reg.fire("on_session_open", "sess_001")
        >>> calls
        ['sess_001']
    """

    @traces(spec="006", story="US-006", reqs=["REQ-006-FR11-001"])
    def __init__(self) -> None:
        """Initialize an empty hook registry with all known hook points."""
        self._hooks: dict[str, list[SessionHook]] = {point: [] for point in HOOK_POINTS}

    @traces(spec="006", story="US-006", reqs=["REQ-006-FR11-001", "REQ-006-US-006-003"])
    def add(self, hook_point: str, callback: SessionHook) -> None:
        """Register a hook callback at a specific hook point.

        Args:
            hook_point: One of ``on_session_open``, ``on_session_close``,
                ``on_fact_emitted``.
            callback: A callable matching the ``SessionHook`` protocol:
                ``callback(session_id: str, **kwargs) -> None``.

        Raises:
            ValueError: If *hook_point* is not a recognised hook point.

        Example:
            >>> reg = HookRegistry()
            >>> reg.add("on_session_open", lambda session_id, **kw: None)
        """
        if hook_point not in HOOK_POINTS:
            raise ValueError(
                f"Unknown hook point {hook_point!r}. "
                f"Valid points: {sorted(HOOK_POINTS)}"
            )
        self._hooks[hook_point].append(callback)

    @traces(spec="006", story="US-006", reqs=["REQ-006-FR11-001"])
    def remove(self, hook_point: str, callback: SessionHook) -> bool:
        """Remove a previously registered hook callback.

        Args:
            hook_point: The hook point the callback was registered under.
            callback: The exact callback object to remove.

        Returns:
            ``True`` if the callback was found and removed, ``False`` otherwise.

        Raises:
            ValueError: If *hook_point* is not a recognised hook point.

        Example:
            >>> reg = HookRegistry()
            >>> fn = lambda session_id, **kw: None
            >>> reg.add("on_session_close", fn)
            >>> reg.remove("on_session_close", fn)
            True
            >>> reg.remove("on_session_close", fn)
            False
        """
        if hook_point not in HOOK_POINTS:
            raise ValueError(
                f"Unknown hook point {hook_point!r}. "
                f"Valid points: {sorted(HOOK_POINTS)}"
            )
        hooks = self._hooks[hook_point]
        try:
            hooks.remove(callback)
            return True
        except ValueError:
            return False

    @traces(
        spec="006",
        story="US-006",
        reqs=[
            "REQ-006-FR11-003",
            "REQ-006-FR11-004",
            "REQ-006-FR11-005",
            "REQ-006-FR6-004",
            "REQ-006-US-006-001",
            "REQ-006-US-006-002",
            "REQ-006-US-006-004",
            "REQ-006-US-006-005",
            "REQ-006-US-006-006",
        ],
    )
    def fire(self, hook_point: str, session_id: str, **kwargs) -> None:
        """Fire all hooks registered at a hook point.

        Each registered callback is invoked with
        ``callback(session_id, **kwargs)``. Exceptions in individual hooks
        are logged but do not prevent subsequent hooks from running.

        Args:
            hook_point: The hook point to fire.
            session_id: The session identifier passed as the first argument
                to every hook.
            **kwargs: Additional keyword arguments forwarded to each hook.

        Raises:
            ValueError: If *hook_point* is not a recognised hook point.

        Example:
            >>> reg = HookRegistry()
            >>> results = []
            >>> reg.add("on_fact_emitted", lambda session_id, **kw: results.append(kw))
            >>> reg.fire("on_fact_emitted", "sess_001", fact_id="f-1", fact_type="decision")
            >>> results
            [{'fact_id': 'f-1', 'fact_type': 'decision'}]
        """
        if hook_point not in HOOK_POINTS:
            raise ValueError(
                f"Unknown hook point {hook_point!r}. "
                f"Valid points: {sorted(HOOK_POINTS)}"
            )
        for hook in self._hooks[hook_point]:
            try:
                hook(session_id, **kwargs)
            except Exception:  # noqa: BLE001 — arbitrary hook, must catch all
                logger.exception(
                    "Hook %r failed at %s for session %s",
                    hook,
                    hook_point,
                    session_id,
                )

    @traces(spec="006", story="FR-011", reqs=["REQ-006-FR11-005"])
    def count(self, hook_point: str) -> int:
        """Return the number of hooks registered at a hook point.

        Args:
            hook_point: The hook point to count.

        Returns:
            Number of registered callbacks.

        Raises:
            ValueError: If *hook_point* is not a recognised hook point.
        """
        if hook_point not in HOOK_POINTS:
            raise ValueError(
                f"Unknown hook point {hook_point!r}. "
                f"Valid points: {sorted(HOOK_POINTS)}"
            )
        return len(self._hooks[hook_point])

    @traces(spec="006", story="US-006", reqs=["REQ-006-FR11-001"])
    def clear(self, hook_point: str | None = None) -> None:
        """Remove all hooks, or all hooks at a specific hook point.

        Args:
            hook_point: If given, clear only that point. If ``None``,
                clear all hook points.

        Raises:
            ValueError: If *hook_point* is given but not recognised.
        """
        if hook_point is not None:
            if hook_point not in HOOK_POINTS:
                raise ValueError(
                    f"Unknown hook point {hook_point!r}. "
                    f"Valid points: {sorted(HOOK_POINTS)}"
                )
            self._hooks[hook_point].clear()
        else:
            for point in HOOK_POINTS:
                self._hooks[point].clear()