Hooks and Metrics API

Observe the transition lifecycle and collect metrics without modifying guards or actions.

See Hooks and Metrics guide for the full guide.

Protocol

TransitionHook

Bases: Protocol

Protocol for transition lifecycle hooks.

Hooks are called in the following order:

1. on_before_process — before any processing (orchestrator level).
2. on_transition_start — a candidate transition is about to fire.
3. on_transition_complete — transition finished (success or fail).
4. on_transition_error — an unhandled exception occurred.

Data classes

TransitionMetrics dataclass

TransitionMetrics(source_state: str, target_state: str, trigger: str, success: bool, duration_ms: float, guards_evaluated: int = 0, guards_passed: int = 0, timestamp: datetime = (lambda: now(UTC))(), machine_name: str = 'unknown', metadata: dict[str, Any] = dict())

Metrics for a single transition.

Built-in hooks

LoggingHook

LoggingHook(log_level: int = DEBUG, logger_name: str = 'pystator.transitions')

Logs transition details.

Parameters:

Name	Type	Description	Default
log_level	int	Logging level for transition messages.	DEBUG
logger_name	str	Name of the logger to use.	'pystator.transitions'

Source code in src/pystator/hooks.py

def __init__(
    self,
    log_level: int = logging.DEBUG,
    logger_name: str = "pystator.transitions",
) -> None:
    """Initialize the logging hook.

    Args:
        log_level: Logging level for transition messages.
        logger_name: Name of the logger to use.
    """
    self.log_level = log_level
    self.logger = logging.getLogger(logger_name)

on_before_process

on_before_process(entity_id: str, current_state: str, trigger: str, context: dict[str, Any]) -> None

Log the start of event processing.

Source code in src/pystator/hooks.py

def on_before_process(
    self,
    entity_id: str,
    current_state: str,
    trigger: str,
    context: dict[str, Any],
) -> None:
    """Log the start of event processing."""
    self.logger.log(
        self.log_level,
        "Processing: entity_id=%s state=%s trigger=%s",
        entity_id,
        current_state,
        trigger,
    )

on_transition_start

on_transition_start(current_state: str, trigger: str, context: dict[str, Any]) -> None

Log the start of a transition.

Source code in src/pystator/hooks.py

def on_transition_start(
    self, current_state: str, trigger: str, context: dict[str, Any]
) -> None:
    """Log the start of a transition."""
    self.logger.log(
        self.log_level,
        "Transition starting: state=%s trigger=%s",
        current_state,
        trigger,
    )

on_transition_complete

on_transition_complete(result: TransitionResult, duration_ms: float, context: dict[str, Any]) -> None

Log the completion of a transition.

Source code in src/pystator/hooks.py

def on_transition_complete(
    self,
    result: TransitionResult,
    duration_ms: float,
    context: dict[str, Any],
) -> None:
    """Log the completion of a transition."""
    if result.success:
        self.logger.log(
            self.log_level,
            "Transition completed: %s -> %s (trigger=%s, duration=%.2fms)",
            result.source_state,
            result.target_state,
            result.trigger,
            duration_ms,
        )
    else:
        self.logger.log(
            self.log_level,
            "Transition failed: state=%s trigger=%s error=%s (duration=%.2fms)",
            result.source_state,
            result.trigger,
            result.error,
            duration_ms,
        )

on_transition_error

on_transition_error(error: Exception, current_state: str, trigger: str, context: dict[str, Any]) -> None

Log a transition error with full traceback.

Source code in src/pystator/hooks.py

def on_transition_error(
    self,
    error: Exception,
    current_state: str,
    trigger: str,
    context: dict[str, Any],
) -> None:
    """Log a transition error with full traceback."""
    self.logger.exception(
        "Transition error: state=%s trigger=%s error=%s",
        current_state,
        trigger,
        error,
    )

MetricsCollector

MetricsCollector(max_history: int = 10000)

Collects and aggregates transition metrics.

Thread-safe with bounded memory. The max_history parameter controls the maximum number of transition records and duration samples kept.

Parameters:

Name	Type	Description	Default
max_history	int	Maximum number of transition records to retain.	10000

Source code in src/pystator/hooks.py

def __init__(self, max_history: int = 10_000) -> None:
    """Initialize the metrics collector.

    Args:
        max_history: Maximum number of transition records to retain.
    """
    self._lock = threading.Lock()
    self._max_history = max_history
    self._transitions: deque[TransitionMetrics] = deque(maxlen=max_history)
    self._counters: dict[str, int] = {
        "total": 0,
        "success": 0,
        "failure": 0,
        "errors": 0,
    }
    self._by_trigger: dict[str, int] = {}
    self._by_source_state: dict[str, int] = {}
    self._by_target_state: dict[str, int] = {}
    self._durations: deque[float] = deque(maxlen=max_history)

on_before_process

on_before_process(entity_id: str, current_state: str, trigger: str, context: dict[str, Any]) -> None

No-op before processing.

Source code in src/pystator/hooks.py

def on_before_process(
    self,
    entity_id: str,
    current_state: str,
    trigger: str,
    context: dict[str, Any],
) -> None:
    """No-op before processing."""
    pass

on_transition_start

on_transition_start(current_state: str, trigger: str, context: dict[str, Any]) -> None

No-op on transition start.

Source code in src/pystator/hooks.py

def on_transition_start(
    self, current_state: str, trigger: str, context: dict[str, Any]
) -> None:
    """No-op on transition start."""
    pass

on_transition_complete

on_transition_complete(result: TransitionResult, duration_ms: float, context: dict[str, Any]) -> None

Record transition metrics on completion.

Source code in src/pystator/hooks.py

def on_transition_complete(
    self,
    result: TransitionResult,
    duration_ms: float,
    context: dict[str, Any],
) -> None:
    """Record transition metrics on completion."""
    with self._lock:
        self._counters["total"] += 1
        if result.success:
            self._counters["success"] += 1
        else:
            self._counters["failure"] += 1
        self._by_trigger[result.trigger] = (
            self._by_trigger.get(result.trigger, 0) + 1
        )
        self._by_source_state[result.source_state] = (
            self._by_source_state.get(result.source_state, 0) + 1
        )
        if result.success and result.target_state:
            self._by_target_state[result.target_state] = (
                self._by_target_state.get(result.target_state, 0) + 1
            )
        self._durations.append(duration_ms)
        self._transitions.append(
            TransitionMetrics(
                source_state=result.source_state,
                target_state=result.target_state or "",
                trigger=result.trigger,
                success=result.success,
                duration_ms=duration_ms,
            )
        )

on_transition_error

on_transition_error(error: Exception, current_state: str, trigger: str, context: dict[str, Any]) -> None

Increment the error counter.

Source code in src/pystator/hooks.py

def on_transition_error(
    self,
    error: Exception,
    current_state: str,
    trigger: str,
    context: dict[str, Any],
) -> None:
    """Increment the error counter."""
    with self._lock:
        self._counters["errors"] += 1

get_summary

get_summary() -> dict[str, Any]

Return aggregated metrics snapshot.

Source code in src/pystator/hooks.py

def get_summary(self) -> dict[str, Any]:
    """Return aggregated metrics snapshot."""
    with self._lock:
        total = self._counters["total"]
        rate = self._counters["success"] / total if total else 0.0
        duration_stats: dict[str, float] = {}
        if self._durations:
            s = sorted(self._durations)
            duration_stats = {
                "min_ms": s[0],
                "max_ms": s[-1],
                "avg_ms": sum(s) / len(s),
                "p50_ms": s[len(s) // 2],
                "p95_ms": s[int(len(s) * 0.95)],
                "p99_ms": s[int(len(s) * 0.99)],
            }
        return {
            "total_transitions": total,
            "successful": self._counters["success"],
            "failed": self._counters["failure"],
            "errors": self._counters["errors"],
            "success_rate": rate,
            "by_trigger": dict(self._by_trigger),
            "by_source_state": dict(self._by_source_state),
            "by_target_state": dict(self._by_target_state),
            "duration": duration_stats,
        }

reset

reset() -> None

Clear all accumulated metrics.

Source code in src/pystator/hooks.py

def reset(self) -> None:
    """Clear all accumulated metrics."""
    with self._lock:
        self._transitions.clear()
        self._counters = {
            "total": 0,
            "success": 0,
            "failure": 0,
            "errors": 0,
        }
        self._by_trigger.clear()
        self._by_source_state.clear()
        self._by_target_state.clear()
        self._durations.clear()

Observer

TransitionObserver

TransitionObserver()

Manages transition lifecycle hooks and provides timing.

Per-thread timing state avoids races when concurrent transitions overwrite each other's start times.

Source code in src/pystator/hooks.py

def __init__(self) -> None:
    """Initialize the observer with empty hook list."""
    self._hooks: list[TransitionHook] = []
    self._local = threading.local()

add_hook

add_hook(hook: TransitionHook) -> TransitionObserver

Register a hook and return self for chaining.

Source code in src/pystator/hooks.py

def add_hook(self, hook: TransitionHook) -> TransitionObserver:
    """Register a hook and return self for chaining."""
    self._hooks.append(hook)
    return self

remove_hook

remove_hook(hook: TransitionHook) -> None

Unregister a previously added hook.

Source code in src/pystator/hooks.py

def remove_hook(self, hook: TransitionHook) -> None:
    """Unregister a previously added hook."""
    self._hooks.remove(hook)

before_process

before_process(entity_id: str, current_state: str, trigger: str, context: dict[str, Any]) -> None

Dispatch on_before_process to all hooks.

Call this at the orchestrator level before any work begins.

Source code in src/pystator/hooks.py

def before_process(
    self,
    entity_id: str,
    current_state: str,
    trigger: str,
    context: dict[str, Any],
) -> None:
    """Dispatch ``on_before_process`` to all hooks.

    Call this at the orchestrator level before any work begins.
    """
    for hook in self._hooks:
        try:
            hook.on_before_process(entity_id, current_state, trigger, context)
        except Exception as e:
            logger.warning("Hook on_before_process failed: %s", e)

before_transition

before_transition(current_state: str, trigger: str, context: dict[str, Any]) -> None

Record start time and dispatch on_transition_start to all hooks.

Source code in src/pystator/hooks.py

def before_transition(
    self, current_state: str, trigger: str, context: dict[str, Any]
) -> None:
    """Record start time and dispatch ``on_transition_start`` to all hooks."""
    self._local.start_time = time.perf_counter()
    self._local.current_state = current_state
    self._local.current_trigger = trigger
    for hook in self._hooks:
        try:
            hook.on_transition_start(current_state, trigger, context)
        except Exception as e:
            logger.warning("Hook on_transition_start failed: %s", e)

after_transition

after_transition(result: TransitionResult, context: dict[str, Any]) -> None

Compute duration and dispatch on_transition_complete to all hooks.

Source code in src/pystator/hooks.py

def after_transition(
    self, result: TransitionResult, context: dict[str, Any]
) -> None:
    """Compute duration and dispatch ``on_transition_complete`` to all hooks."""
    duration_ms = 0.0
    start = getattr(self._local, "start_time", None)
    if start is not None:
        duration_ms = (time.perf_counter() - start) * 1000
        self._local.start_time = None
    for hook in self._hooks:
        try:
            hook.on_transition_complete(result, duration_ms, context)
        except Exception as e:
            logger.warning("Hook on_transition_complete failed: %s", e)

on_error

on_error(error: Exception, context: dict[str, Any]) -> None

Dispatch on_transition_error to all hooks.

Source code in src/pystator/hooks.py

def on_error(self, error: Exception, context: dict[str, Any]) -> None:
    """Dispatch ``on_transition_error`` to all hooks."""
    state = getattr(self._local, "current_state", None) or "unknown"
    trigger = getattr(self._local, "current_trigger", None) or "unknown"
    for hook in self._hooks:
        try:
            hook.on_transition_error(error, state, trigger, context)
        except Exception as e:
            logger.warning("Hook on_transition_error failed: %s", e)

Import

from pystator.hooks import (
    TransitionHook,
    TransitionMetrics,
    LoggingHook,
    MetricsCollector,
    TransitionObserver,
)

See also

• Hooks and Metrics guide
• Orchestrator — hooks parameter