Events

The event system lets you observe graph execution without modifying your workflow logic. Events are emitted at key points — run start/end, node start/end, routing decisions — and delivered to processors you provide.

• Event types - Immutable dataclasses for each execution milestone

• EventProcessor - Sync base class for consuming events

• AsyncEventProcessor - Async-aware variant for async runners

• TypedEventProcessor - Auto-dispatches to typed handler methods

• RichProgressProcessor - Hierarchical Rich progress bars out of the box

Overview

from hypergraph import SyncRunner, RichProgressProcessor

runner = SyncRunner()
result = runner.run(graph, inputs, event_processors=[RichProgressProcessor()])

Events flow through this pipeline:

Runner emits event → EventDispatcher → each EventProcessor.on_event()

All dispatch is best-effort: a failing processor never breaks execution.

---

Event Types

All events inherit from BaseEvent and are frozen dataclasses (immutable after creation).

BaseEvent

@dataclass(frozen=True)
class BaseEvent:
    run_id: str                    # Unique identifier for the run
    span_id: str                   # Unique identifier for this event's scope
    parent_span_id: str | None     # Parent scope, or None for root runs
    workflow_id: str | None        # Persisted workflow identifier, if any
    item_index: int | None         # Mapped child index, if any
    timestamp: float               # Unix timestamp

Every event carries tracing context: run_id groups events from the same execution, span_id uniquely identifies the scope, and parent_span_id links nested graphs to their parents.

RunStartEvent

Emitted when a graph run begins.

@dataclass(frozen=True)
class RunStartEvent(BaseEvent):
    graph_name: str              # Name of the graph
    is_map: bool                 # True if this is a map() operation
    map_size: int | None         # Number of items in map, if applicable
    parent_workflow_id: str | None
    forked_from: str | None
    fork_superstep: int | None
    retry_of: str | None
    retry_index: int | None
    is_resume: bool

RunEndEvent

Emitted when a graph run completes (successfully or not).

@dataclass(frozen=True)
class RunEndEvent(BaseEvent):
    graph_name: str              # Name of the graph
    status: str                  # "completed", "failed", "paused", "partial", or "stopped"
    error: str | None            # Error message if failed
    duration_ms: float           # Wall-clock duration in milliseconds
    batch_total_items: int | None
    batch_completed_items: int | None
    batch_failed_items: int | None
    batch_paused_items: int | None
    batch_stopped_items: int | None
    batch_outcome: str | None

NodeStartEvent

Emitted when a node begins execution.

@dataclass(frozen=True)
class NodeStartEvent(BaseEvent):
    node_name: str               # Name of the node
    graph_name: str              # Graph containing the node
    superstep: int | None        # Zero-indexed superstep, if known

NodeEndEvent

Emitted when a node completes successfully.

@dataclass(frozen=True)
class NodeEndEvent(BaseEvent):
    node_name: str               # Name of the node
    graph_name: str              # Graph containing the node
    superstep: int | None        # Zero-indexed superstep, if known
    duration_ms: float           # Wall-clock duration in milliseconds
    cached: bool                 # True if result was served from cache

NodeErrorEvent

Emitted when a node fails with an exception.

@dataclass(frozen=True)
class NodeErrorEvent(BaseEvent):
    node_name: str               # Name of the node
    graph_name: str              # Graph containing the node
    error: str                   # Error message
    error_type: str              # Fully qualified exception type
    superstep: int | None        # Zero-indexed superstep, if known

RouteDecisionEvent

Emitted when a routing node (@route or @ifelse) makes a decision.

@dataclass(frozen=True)
class RouteDecisionEvent(BaseEvent):
    node_name: str               # Name of the routing node
    graph_name: str              # Graph containing the node
    decision: str | list[str]    # Chosen target(s)
    node_span_id: str | None     # Routing node span, when available
    superstep: int | None        # Zero-indexed superstep, if known

InterruptEvent

Emitted when execution pauses for human-in-the-loop input.

@dataclass(frozen=True)
class InterruptEvent(BaseEvent):
    node_name: str               # Node that triggered the interrupt
    graph_name: str              # Graph containing the node
    value: object                # Interrupt payload
    response_param: str          # Parameter name for the response
    superstep: int | None        # Zero-indexed superstep, if known

StopRequestedEvent

Emitted when a stop is requested on a workflow.

@dataclass(frozen=True)
class StopRequestedEvent(BaseEvent):
    graph_name: str              # Graph being stopped
    info: object                 # Optional stop metadata

CacheHitEvent

Emitted when a node result is served from cache instead of being executed.

@dataclass(frozen=True)
class CacheHitEvent(BaseEvent):
    node_name: str               # Name of the cached node
    graph_name: str              # Graph containing the node
    cache_key: str               # The cache key that was hit
    superstep: int | None        # Zero-indexed superstep, if known

InnerCacheEvent

Emitted when a hypercache-decorated call happens inside a running node. This bridge activates when hypercache is installed, including via pip install 'hypergraph[cache]'. The event is emitted when the installed Hypercache version exposes its public observer API.

@dataclass(frozen=True)
class InnerCacheEvent(BaseEvent):
    node_name: str               # Name of the graph node that triggered the call
    graph_name: str              # Graph containing the node
    instance: str                # Qualified cache instance name
    operation: str               # Cached method name
    hit: bool                    # True if served from cache
    stale: bool                  # True if the cached value was stale
    refreshing: bool             # True if background refresh was triggered
    wrote: bool                  # True if a new value was written
    mode: str                    # Cache mode in effect

Event (Union Type)

Event = (
    RunStartEvent | RunEndEvent | NodeStartEvent | NodeEndEvent
    | NodeErrorEvent | RouteDecisionEvent | InterruptEvent | StopRequestedEvent
    | CacheHitEvent | StreamingChunkEvent | InnerCacheEvent
)

---

Processor Interfaces

EventProcessor

Base class for synchronous event consumers.

class EventProcessor:
    def on_event(self, event: Event) -> None:
        """Called for every event."""

    def shutdown(self) -> None:
        """Called once when the run completes. Flush buffers here."""

Example — logging processor:

from hypergraph import EventProcessor

class LoggingProcessor(EventProcessor):
    def on_event(self, event):
        print(f"[{type(event).__name__}] {event.span_id}")

AsyncEventProcessor

Extends EventProcessor with async variants. The async runner prefers on_event_async and shutdown_async when available, falling back to sync methods otherwise.

class AsyncEventProcessor(EventProcessor):
    async def on_event_async(self, event: Event) -> None:
        """Async version of on_event."""

    async def shutdown_async(self) -> None:
        """Async version of shutdown."""

TypedEventProcessor

Auto-dispatches on_event to typed handler methods. Override only the events you care about — unhandled types are silently ignored.

class TypedEventProcessor(EventProcessor):
    def on_run_start(self, event: RunStartEvent) -> None: ...
    def on_run_end(self, event: RunEndEvent) -> None: ...
    def on_node_start(self, event: NodeStartEvent) -> None: ...
    def on_node_end(self, event: NodeEndEvent) -> None: ...
    def on_node_error(self, event: NodeErrorEvent) -> None: ...
    def on_route_decision(self, event: RouteDecisionEvent) -> None: ...
    def on_interrupt(self, event: InterruptEvent) -> None: ...
    def on_stop_requested(self, event: StopRequestedEvent) -> None: ...
    def on_cache_hit(self, event: CacheHitEvent) -> None: ...
    def on_streaming_chunk(self, event: StreamingChunkEvent) -> None: ...
    def on_inner_cache(self, event: InnerCacheEvent) -> None: ...

Example — timing processor:

from hypergraph import TypedEventProcessor, NodeEndEvent

class SlowNodeDetector(TypedEventProcessor):
    def on_node_end(self, event: NodeEndEvent) -> None:
        if event.duration_ms > 1000:
            print(f"⚠️  Slow node: {event.node_name} ({event.duration_ms:.0f}ms)")

---

RichProgressProcessor

Hierarchical Rich progress bars for graph execution. Requires the rich package.

pip install 'hypergraph[progress]'
# or
pip install rich

Usage

from hypergraph import SyncRunner, RichProgressProcessor

runner = SyncRunner()
result = runner.run(graph, inputs, event_processors=[RichProgressProcessor()])

Constructor

class RichProgressProcessor(TypedEventProcessor):
    def __init__(
        self,
        *,
        transient: bool = True,
        force_mode: Literal["tty", "non-tty", "auto"] = "auto",
    ) -> None: ...

Args:

• transient - If True (default), progress bars are removed after completion. Set to False to keep them visible.

• force_mode - Controls output mode:

  • "auto" (default): detect via stdout.isatty()

  • "tty": force Rich live bars

  • "non-tty": force plain-text milestone logging (useful for CI/log pipelines)

Visual Output

Single run:

📦 my_graph ━━━━━━━━━━━━━━━━━━━━ 100% 3/3

Nested graph:

📦 outer_graph ━━━━━━━━━━━━━━━━━ 100% 3/3
  🌳 inner_rag ━━━━━━━━━━━━━━━━━ 100% 2/2

Map operation:

🗺️ scrape_graph Progress ━━━━━━━ 100% 50/50
  📦 fetch ━━━━━━━━━━━━━━━━━━━━━ 100% 50/50
  📦 parse ━━━━━━━━━━━━━━━━━━━━━ 100% 50/50

Map with failures:

🗺️ scrape_graph Progress ━━━━━━━ 100% 50/50 (3 failed)
  📦 fetch ━━━━━━━━━━━━━━━━━━━━━ 100% 50/50
  📦 parse ━━━━━━━━━━━━━━━━━━━━━  94% 47/50

Non-TTY Milestones

In non-TTY mode, map progress is logged at fixed milestones (10%, 25%, 50%, 75%, 100%) instead of rendering live bars:

[14:20:00] 🗺️ scrape_graph: 50% (50/100)

Visual Conventions

Icon	Meaning
📦	Regular node (depth 0)
🌳	Nested graph node (depth > 0)
🗺️	Map-level progress bar

Indentation reflects nesting depth. Failed nodes show [red]FAILED[/red] in the description.

---

EventDispatcher

Manages processor lifecycle and event delivery. You typically don't interact with this directly — runners create it internally from event_processors.

class EventDispatcher:
    def __init__(self, processors: list[EventProcessor] | None = None) -> None: ...

    @property
    def active(self) -> bool:
        """True if there is at least one registered processor."""

    def emit(self, event: Event) -> None:
        """Send event to every processor synchronously."""

    async def emit_async(self, event: Event) -> None:
        """Send event to every processor, using async when available."""

    def shutdown(self) -> None:
        """Shut down all processors. Best-effort."""

    async def shutdown_async(self) -> None:
        """Shut down all processors, using async when available."""

---

Event Sequence

A typical DAG run emits events in this order:

RunStartEvent(graph_name="rag")
  NodeStartEvent(node_name="embed")
  NodeEndEvent(node_name="embed")
  NodeStartEvent(node_name="retrieve")
  NodeEndEvent(node_name="retrieve")
  NodeStartEvent(node_name="generate")
  NodeEndEvent(node_name="generate")
RunEndEvent(graph_name="rag", status="completed")

For cached nodes, the sequence includes a CacheHitEvent:

NodeStartEvent(node_name="embed")
CacheHitEvent(node_name="embed", cache_key="abc123...")
NodeEndEvent(node_name="embed", cached=True, duration_ms=0.0)

For map operations, each item gets its own RunStartEvent/RunEndEvent pair nested under the map's span:

RunStartEvent(is_map=True, map_size=3)
  RunStartEvent(graph_name="pipeline")   # item 1
    NodeStartEvent / NodeEndEvent ...
  RunEndEvent(graph_name="pipeline")
  RunStartEvent(graph_name="pipeline")   # item 2
    ...
  RunEndEvent(graph_name="pipeline")
  ...
RunEndEvent(status="completed")

For nested graphs, parent_span_id links inner events to the outer run:

RunStartEvent(graph_name="outer")
  NodeStartEvent(node_name="validate")
  NodeEndEvent(node_name="validate")
  NodeStartEvent(node_name="rag")        # GraphNode start
    RunStartEvent(graph_name="rag", parent_span_id=<outer_span>)
      NodeStartEvent(node_name="embed")
      NodeEndEvent(node_name="embed")
      ...
    RunEndEvent(graph_name="rag")
  NodeEndEvent(node_name="rag")          # GraphNode end
RunEndEvent(graph_name="outer")