Skip to content

forge.events

Decoupled, asynchronous module communication via an in-process event bus.

forge.events

Event bus module — decoupled, asynchronous module communication.

Provides an in-process event emitter and subscriber system for modules to communicate without direct dependencies. Supports typed events, async handlers, lifecycle event hooks, wildcard pattern subscriptions, and an event history buffer for late-joining subscribers.

Attributes

Handler module-attribute 

Handler = Callable[..., Awaitable[Any]]

Signature that every event handler must satisfy.

Classes

EmptyPayload

Bases: TypedDict

Events that carry no data beyond the event name.

Source code in src/forge/events/event_types.py 
37
38
class EmptyPayload(TypedDict):
    """Events that carry no data beyond the event name."""

EventBusWrapper

Wraps the core :class:~forge.core.events.EventBus with wildcard pattern support and an event history buffer.

Every method is compatible with the core EventBus interface so this class acts as a drop-in replacement for runtime._events.

Source code in src/forge/events/module.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
class EventBusWrapper:
    """
    Wraps the core :class:`~forge.core.events.EventBus` with wildcard
    pattern support and an event history buffer.

    Every method is compatible with the core ``EventBus`` interface so
    this class acts as a drop-in replacement for ``runtime._events``.
    """

    def __init__(self, core_bus: EventBus, history_size: int = 200) -> None:
        self._core: EventBus = core_bus
        self._history: deque[EventRecord] = deque(maxlen=history_size)
        self._wildcard_handlers: dict[str, set[Handler]] = {}

    # -- Public registration / unregistration ---------------------------------

    def on(
        self,
        event: str,
        handler: Handler | None = None,
    ) -> Handler | Callable[[Handler], Handler]:
        """
        Register an async handler for *event*.

        Supports three calling conventions:

        * Direct: ``bus.on("event.name", handler)``
        * Decorator: ``@bus.on("event.name")``
        * Wildcard: ``bus.on("ai.*", handler)`` or ``@bus.on("*")``

        Wildcard patterns use shell-style globbing (``fnmatch``).
        """

        def _register(h: Handler) -> Handler:
            if _is_wildcard(event):
                self._wildcard_handlers.setdefault(event, set()).add(h)
            else:
                self._core.on(event, h)
            return h

        if handler is not None:
            return _register(handler)
        return _register

    def off(self, event: str, handler: Handler) -> None:
        """
        Unregister a previously registered handler.

        Raises ``KeyError`` if the handler is not registered for the
        given *event* (or wildcard pattern).
        """
        if _is_wildcard(event):
            handlers = self._wildcard_handlers.get(event)
            if handlers is None or handler not in handlers:
                msg = f"Handler {handler!r} is not registered for pattern {event!r}."
                raise KeyError(msg)
            handlers.discard(handler)
            if not handlers:
                del self._wildcard_handlers[event]
        else:
            self._core.off(event, handler)

    # -- Subscription with optional history replay ----------------------------

    def subscribe(
        self,
        event: str,
        handler: Handler,
        *,
        replay: bool = False,
    ) -> Handler:
        """
        Register *handler* for *event*, optionally replaying matching
        history entries to the handler immediately.

        This is the primary API for late-joining subscribers::

            @runtime.events.subscribe("ai.*", replay=True)
            async def track_ai(**kwargs):
                ...
        """
        self.on(event, handler)
        if replay:
            self.replay(event, handler)
        return handler

    def replay(self, event: str, handler: Handler) -> None:
        """Fire *handler* for every history entry that matches *event*."""
        _tasks: list[asyncio.Task[None]] = [
            asyncio.ensure_future(handler(**record.payload))
            for record in self._history
            if fnmatch.fnmatch(record.event, event)
        ]

    # -- Emission -------------------------------------------------------------

    async def emit(self, event: str, **kwargs: Any) -> None:
        """
        Emit *event* and await all registered handlers concurrently.

        Stores the event in the history buffer, dispatches exact-match
        handlers via the core bus, and then dispatches wildcard handlers
        that match *event*.

        Exceptions from individual handlers are caught and logged so
        that a failing handler never blocks other subscribers.
        """
        record = EventRecord(
            event=event,
            payload=kwargs,
            timestamp=time.monotonic(),
        )
        self._history.append(record)

        await self._core.emit(event, **kwargs)

        wildcard_tasks: list[asyncio.Task[None]] = []
        for pattern, handlers in self._wildcard_handlers.items():
            if fnmatch.fnmatch(event, pattern):
                wildcard_tasks.extend(
                    asyncio.create_task(
                        self._safe_dispatch(handler, event, kwargs),
                    )
                    for handler in handlers
                )

        if wildcard_tasks:
            results = await asyncio.gather(*wildcard_tasks, return_exceptions=True)
            for exc in results:
                if isinstance(exc, BaseException):
                    _log.warning(
                        "Wildcard handler for '%s' raised %s: %s",
                        event,
                        type(exc).__name__,
                        exc,
                    )

    # -- Introspection --------------------------------------------------------

    def has_listeners(self, event: str) -> bool:
        """Return ``True`` if at least one handler is registered for *event*."""
        if self._core.has_listeners(event):
            return True
        return any(fnmatch.fnmatch(event, p) for p in self._wildcard_handlers)

    @property
    def history(self) -> tuple[EventRecord, ...]:
        """Immutable snapshot of the event history buffer."""
        return tuple(self._history)

    def clear_history(self) -> None:
        """Discard all stored events from the history buffer."""
        self._history.clear()

    # -- Internal helpers -----------------------------------------------------

    async def _safe_dispatch(
        self,
        handler: Handler,
        event: str,
        payload: dict[str, Any],
    ) -> None:
        try:
            await handler(**payload)
        except Exception:
            _log.exception("Unhandled exception in wildcard handler for '%s'", event)
            raise
Attributes
history property 
history: tuple[EventRecord, ...]

Immutable snapshot of the event history buffer.

Methods:
clear_history 
clear_history() -> None

Discard all stored events from the history buffer.

Source code in src/forge/events/module.py 
228
229
230
def clear_history(self) -> None:
    """Discard all stored events from the history buffer."""
    self._history.clear()
emit async 
emit(event: str, **kwargs: Any) -> None

Emit event and await all registered handlers concurrently.

Stores the event in the history buffer, dispatches exact-match handlers via the core bus, and then dispatches wildcard handlers that match event.

Exceptions from individual handlers are caught and logged so that a failing handler never blocks other subscribers.

Source code in src/forge/events/module.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
async def emit(self, event: str, **kwargs: Any) -> None:
    """
    Emit *event* and await all registered handlers concurrently.

    Stores the event in the history buffer, dispatches exact-match
    handlers via the core bus, and then dispatches wildcard handlers
    that match *event*.

    Exceptions from individual handlers are caught and logged so
    that a failing handler never blocks other subscribers.
    """
    record = EventRecord(
        event=event,
        payload=kwargs,
        timestamp=time.monotonic(),
    )
    self._history.append(record)

    await self._core.emit(event, **kwargs)

    wildcard_tasks: list[asyncio.Task[None]] = []
    for pattern, handlers in self._wildcard_handlers.items():
        if fnmatch.fnmatch(event, pattern):
            wildcard_tasks.extend(
                asyncio.create_task(
                    self._safe_dispatch(handler, event, kwargs),
                )
                for handler in handlers
            )

    if wildcard_tasks:
        results = await asyncio.gather(*wildcard_tasks, return_exceptions=True)
        for exc in results:
            if isinstance(exc, BaseException):
                _log.warning(
                    "Wildcard handler for '%s' raised %s: %s",
                    event,
                    type(exc).__name__,
                    exc,
                )
has_listeners 
has_listeners(event: str) -> bool

Return True if at least one handler is registered for event.

Source code in src/forge/events/module.py 
217
218
219
220
221
def has_listeners(self, event: str) -> bool:
    """Return ``True`` if at least one handler is registered for *event*."""
    if self._core.has_listeners(event):
        return True
    return any(fnmatch.fnmatch(event, p) for p in self._wildcard_handlers)
off 
off(event: str, handler: Handler) -> None

Unregister a previously registered handler.

Raises KeyError if the handler is not registered for the given event (or wildcard pattern).

Source code in src/forge/events/module.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def off(self, event: str, handler: Handler) -> None:
    """
    Unregister a previously registered handler.

    Raises ``KeyError`` if the handler is not registered for the
    given *event* (or wildcard pattern).
    """
    if _is_wildcard(event):
        handlers = self._wildcard_handlers.get(event)
        if handlers is None or handler not in handlers:
            msg = f"Handler {handler!r} is not registered for pattern {event!r}."
            raise KeyError(msg)
        handlers.discard(handler)
        if not handlers:
            del self._wildcard_handlers[event]
    else:
        self._core.off(event, handler)
on 
on(event: str, handler: Handler | None = None) -> Handler | Callable[[Handler], Handler]

Register an async handler for event.

Supports three calling conventions:

  • Direct: bus.on("event.name", handler)
  • Decorator: @bus.on("event.name")
  • Wildcard: bus.on("ai.*", handler) or @bus.on("*")

Wildcard patterns use shell-style globbing (fnmatch).

Source code in src/forge/events/module.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def on(
    self,
    event: str,
    handler: Handler | None = None,
) -> Handler | Callable[[Handler], Handler]:
    """
    Register an async handler for *event*.

    Supports three calling conventions:

    * Direct: ``bus.on("event.name", handler)``
    * Decorator: ``@bus.on("event.name")``
    * Wildcard: ``bus.on("ai.*", handler)`` or ``@bus.on("*")``

    Wildcard patterns use shell-style globbing (``fnmatch``).
    """

    def _register(h: Handler) -> Handler:
        if _is_wildcard(event):
            self._wildcard_handlers.setdefault(event, set()).add(h)
        else:
            self._core.on(event, h)
        return h

    if handler is not None:
        return _register(handler)
    return _register
replay 
replay(event: str, handler: Handler) -> None

Fire handler for every history entry that matches event.

Source code in src/forge/events/module.py 
164
165
166
167
168
169
170
def replay(self, event: str, handler: Handler) -> None:
    """Fire *handler* for every history entry that matches *event*."""
    _tasks: list[asyncio.Task[None]] = [
        asyncio.ensure_future(handler(**record.payload))
        for record in self._history
        if fnmatch.fnmatch(record.event, event)
    ]
subscribe 
subscribe(event: str, handler: Handler, *, replay: bool = False) -> Handler

Register handler for event, optionally replaying matching history entries to the handler immediately.

This is the primary API for late-joining subscribers::

@runtime.events.subscribe("ai.*", replay=True)
async def track_ai(**kwargs):
    ...
Source code in src/forge/events/module.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def subscribe(
    self,
    event: str,
    handler: Handler,
    *,
    replay: bool = False,
) -> Handler:
    """
    Register *handler* for *event*, optionally replaying matching
    history entries to the handler immediately.

    This is the primary API for late-joining subscribers::

        @runtime.events.subscribe("ai.*", replay=True)
        async def track_ai(**kwargs):
            ...
    """
    self.on(event, handler)
    if replay:
        self.replay(event, handler)
    return handler

EventRecord dataclass

A single entry in the event history buffer.

Source code in src/forge/events/module.py 
64
65
66
67
68
69
70
@dataclass
class EventRecord:
    """A single entry in the event history buffer."""

    event: str
    payload: dict[str, Any]
    timestamp: float

EventsModule

Bases: ForgeModule

Forge module that upgrades the core EventBus with wildcard subscriptions, history buffering, and late-joiner replay.

Register this module with the runtime early so that every other module can benefit from the enhanced event API::

runtime.register(EventsModule(history_size=500))
Source code in src/forge/events/module.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
class EventsModule(ForgeModule):
    """
    Forge module that upgrades the core ``EventBus`` with wildcard
    subscriptions, history buffering, and late-joiner replay.

    Register this module with the runtime early so that every other
    module can benefit from the enhanced event API::

        runtime.register(EventsModule(history_size=500))
    """

    name = "events"
    dependencies: ClassVar[list[str]] = []

    def __init__(self, history_size: int = 200) -> None:
        super().__init__()
        self._wrapper: EventBusWrapper | None = None
        self._runtime: Runtime | None = None
        self._core_bus: EventBus | None = None
        self._history_size = history_size

    # -- Public accessor ------------------------------------------------------

    @property
    def bus(self) -> EventBusWrapper:
        """The :class:`EventBusWrapper` instance installed on the runtime."""
        if self._wrapper is None:
            raise RuntimeError(
                "EventsModule is not initialized. "
                "Register it with the runtime and call await runtime.init().",
            )
        return self._wrapper

    # -- Lifecycle ------------------------------------------------------------

    async def setup(self, runtime: Runtime) -> None:
        self._runtime = runtime
        self._core_bus = runtime._events
        if self._core_bus is None:
            from forge.core.exceptions import ForgeError

            raise ForgeError("Runtime does not have an _events attribute.")

        self._wrapper = EventBusWrapper(
            core_bus=self._core_bus,
            history_size=self._history_size,
        )
        runtime._events = self._wrapper  # type: ignore[assignment]

    async def teardown(self) -> None:
        if self._wrapper is not None:
            self._wrapper.clear_history()
        self._wrapper = None
        self._runtime = None

    # -- Health ---------------------------------------------------------------

    def health_check(self) -> HealthResult:
        if self._wrapper is None:
            return HealthResult.error("EventsModule is not initialized.")
        return HealthResult.ok()
Attributes
bus property 
bus: EventBusWrapper

The :class:EventBusWrapper instance installed on the runtime.