Hooks and Metrics

Hooks let you observe the transition lifecycle — attach logging, collect metrics, trigger alerts, or integrate with tracing systems — without modifying your guards or actions.

The TransitionHook protocol

Any object that implements the four methods below can be registered as a hook:

class TransitionHook(Protocol):
    def on_before_process(
        self, entity_id: str, current_state: str, trigger: str, context: dict
    ) -> None: ...

    def on_transition_start(
        self, current_state: str, trigger: str, context: dict
    ) -> None: ...

    def on_transition_complete(
        self, result: TransitionResult, duration_ms: float, context: dict
    ) -> None: ...

    def on_transition_error(
        self, error: Exception, current_state: str, trigger: str, context: dict
    ) -> None: ...

Method	When called
on_before_process	Orchestrator level — before any guard evaluation
on_transition_start	A candidate transition is about to fire
on_transition_complete	Transition finished (success or failure)
on_transition_error	An unhandled exception occurred

---

Built-in hooks

LoggingHook

Logs every transition lifecycle event using Python's standard logging:

import logging
from pystator import Orchestrator
from pystator.hooks import LoggingHook

orchestrator = Orchestrator(
    machine=machine,
    store=store,
    hooks=[LoggingHook(log_level=logging.INFO)],
)

Default logger name: pystator.transitions. Override it:

hook = LoggingHook(
    log_level=logging.DEBUG,
    logger_name="myapp.fsm.transitions",
)

MetricsCollector

Aggregates transition counts, success rates, latency percentiles, and per-trigger / per-state breakdowns:

from pystator import Orchestrator
from pystator.hooks import MetricsCollector

collector = MetricsCollector()
orchestrator = Orchestrator(machine=machine, store=store, hooks=[collector])

# ... process many events ...

summary = collector.get_summary()
print(summary)

get_summary() return keys:

Key	Type	Description
total_transitions	int	Total transitions attempted
successful	int	Transitions that completed with success=True
failed	int	Transitions that completed with success=False
errors	int	Transitions that raised an exception
success_rate	float	successful / total_transitions
by_trigger	dict[str, int]	Count per trigger name
by_source_state	dict[str, int]	Count per source state
by_target_state	dict[str, int]	Count per target state
duration.min_ms	float	Minimum transition latency (ms)
duration.max_ms	float	Maximum transition latency (ms)
duration.avg_ms	float	Average latency
duration.p50_ms	float	Median latency
duration.p95_ms	float	95th percentile latency
duration.p99_ms	float	99th percentile latency

Reset metrics between reporting windows:

# Emit metrics to your monitoring system
emit_to_prometheus(collector.get_summary())

# Start fresh for the next window
collector.reset()

---

Writing a custom hook

Any class with the four methods works — you do not need to inherit from anything:

from pystator.hooks import TransitionMetrics

class DatadogHook:
    def __init__(self, dd_client):
        self._dd = dd_client

    def on_before_process(self, entity_id, current_state, trigger, context):
        pass  # no-op

    def on_transition_start(self, current_state, trigger, context):
        pass

    def on_transition_complete(self, result, duration_ms, context):
        tags = [f"trigger:{result.trigger}", f"state:{result.source_state}"]
        self._dd.timing("fsm.transition.duration_ms", duration_ms, tags=tags)
        if result.success:
            self._dd.increment("fsm.transition.success", tags=tags)
        else:
            self._dd.increment("fsm.transition.failure", tags=tags)

    def on_transition_error(self, error, current_state, trigger, context):
        self._dd.increment("fsm.transition.error", tags=[f"trigger:{trigger}"])

Pass it in alongside the built-in hooks:

orchestrator = Orchestrator(
    machine=machine,
    store=store,
    hooks=[LoggingHook(), MetricsCollector(), DatadogHook(dd)],
)

---

Combining hooks with the TransitionObserver

TransitionObserver manages a list of hooks internally and handles timing. You typically don't use it directly — it's used inside Orchestrator — but it is useful when you manage hook lifecycles manually:

from pystator.hooks import TransitionObserver, LoggingHook, MetricsCollector

observer = TransitionObserver()
observer.add_hook(LoggingHook())

collector = MetricsCollector()
observer.add_hook(collector)

---

OpenTelemetry tracing

Implement a hook that creates spans:

from opentelemetry import trace

tracer = trace.get_tracer("pystator")

class TracingHook:
    def __init__(self):
        self._span = None

    def on_before_process(self, entity_id, current_state, trigger, context):
        self._span = tracer.start_span(
            f"fsm.process",
            attributes={
                "entity.id": entity_id,
                "fsm.state": current_state,
                "fsm.trigger": trigger,
            },
        )

    def on_transition_start(self, current_state, trigger, context):
        pass

    def on_transition_complete(self, result, duration_ms, context):
        if self._span:
            self._span.set_attribute("fsm.success", result.success)
            if result.success:
                self._span.set_attribute("fsm.target_state", result.target_state or "")
            self._span.end()

    def on_transition_error(self, error, current_state, trigger, context):
        if self._span:
            self._span.record_exception(error)
            self._span.end()

---

Prometheus integration

from prometheus_client import Counter, Histogram
from pystator.hooks import MetricsCollector

transitions_total = Counter(
    "fsm_transitions_total", "Total transitions", ["trigger", "source_state", "success"]
)
transition_duration = Histogram(
    "fsm_transition_duration_ms", "Transition duration", ["trigger"]
)

class PrometheusHook:
    def on_before_process(self, *args): pass
    def on_transition_start(self, *args): pass

    def on_transition_complete(self, result, duration_ms, context):
        transitions_total.labels(
            trigger=result.trigger,
            source_state=result.source_state,
            success=str(result.success),
        ).inc()
        transition_duration.labels(trigger=result.trigger).observe(duration_ms)

    def on_transition_error(self, error, current_state, trigger, context):
        transitions_total.labels(
            trigger=trigger, source_state=current_state, success="False"
        ).inc()

---

See also

• Orchestrator API — hooks parameter
• API Reference — Hooks
• Deployment — Prometheus and OpenTelemetry in production