How the Event Bus Handles Real-Time Communication Between Agent-Native Components

Agent-Native uses a lightweight, in-process typed pub/sub bus that propagates events instantly across the framework via a global singleton, Zod-validated payloads, and isolated dispatch semantics.

The BuilderIO/agent-native repository implements a high-performance event bus that enables real-time communication between frontend UI components, server actions, plugins, and background tasks without network overhead. This system relies on a global singleton pattern and strict type validation to ensure reliable, instantaneous message passing across all parts of the framework.

Global Singleton Pattern

The event bus lives as a global singleton on globalThis under the well-known symbol @agent-native/core/event-bus.bus. The getBus() function lazily creates an EventEmitter and a Map of subscription records, guaranteeing that every module—whether running in the Vite dev server, Nitro server, or built bundle—shares the same event hub.

In packages/core/src/event-bus/bus.ts (lines 22-37), the singleton initialization ensures that emissions from any component immediately reach all subscribed listeners across the entire process.

Typed Payload Validation

Before an event is emitted, the bus performs strict type checking. The registerEvent function stores event definitions in a global registry (packages/core/src/event-bus/registry.ts, lines 23-48), including a Zod schema in payloadSchema.

When emit is called (packages/core/src/event-bus/bus.ts, lines 70-90), the bus looks up the registered definition and validates the payload against the schema. Successful validation yields a coerced value passed to listeners; failed validation aborts the emission and logs a warning. This guarantees that real-time communication remains type-safe across loosely coupled components.

Subscription Lifecycle Management

The subscription API provides precise control over listener registration. The subscribe(event, handler) function:

1. Generates a unique UUID using randomUUID()
2. Stores the handler tuple in the subscriptions Map
3. Attaches the handler to the underlying EventEmitter

The unsubscribe(id) function safely removes the listener and its record. This design enables precise clean-up when components unmount or actions complete, preventing memory leaks in long-running processes.

These mechanisms are implemented in packages/core/src/event-bus/bus.ts (lines 39-60).

Real-Time Dispatch Semantics

When emit is called, the bus guarantees isolated and ordered delivery through three specific mechanisms:

Snapshot Isolation – The current listeners are snap-shotted using bus.emitter.listeners(event) before dispatch begins. This ensures that listeners added or removed during emission do not affect the ongoing dispatch.

Error Boundaries – Each listener runs inside a try/catch block. Synchronous exceptions are logged but do not stop later listeners. Asynchronous rejections are caught with .catch() and logged, preventing unhandled promise rejections from breaking the event chain.

Ordered Execution – Handlers execute in the order they were subscribed, maintaining predictable behavior for dependent operations.

These semantics are defined in packages/core/src/event-bus/bus.ts (lines 101-120).

Event Metadata and Tracing

Every emission automatically receives an EventMeta object containing a unique eventId, an ISO-formatted emittedAt timestamp, and optional user-provided fields such as owner. Handlers receive both the validated payload and this metadata, enabling distributed tracing and debugging of real-time flows across the system.

This metadata construction occurs in packages/core/src/event-bus/bus.ts (lines 96-102).

Built-in Event Registry

The registry seeds core events—such as agent.turn.completed—on startup (packages/core/src/event-bus/registry.ts, lines 58-78), ensuring the framework can react to critical lifecycle moments. Extensions register additional events at module load time, instantly making them available to any component without configuration overhead.

Practical Implementation Examples

Register a custom event type at module load time:

import { registerEvent } from "@agent-native/core/event-bus/registry";
import { z } from "zod";

registerEvent({
  name: "user.profile.updated",
  description: "Fired when a user updates their profile",
  payloadSchema: z.object({
    userId: z.string(),
    changes: z.record(z.string(), z.unknown()),
  }),
});

Subscribe from any component:

import { subscribe } from "@agent-native/core/event-bus/bus";

const subId = subscribe("user.profile.updated", (payload, meta) => {
  console.log("Profile update:", payload);
  console.log("Origin:", meta.owner);
});

Emit the event with metadata:

import { emit } from "@agent-native/core/event-bus/bus";

await emit(
  "user.profile.updated",
  { userId: "u123", changes: { displayName: "New Name" } },
  { owner: "admin@example.com" },
);

Clean up subscriptions to prevent memory leaks:

import { unsubscribe } from "@agent-native/core/event-bus/bus";

const ok = unsubscribe(subId);   // true if the subscription existed

Debug active subscriptions:

import { listSubscriptions } from "@agent-native/core/event-bus/bus";

console.table(listSubscriptions("user.profile.updated"));

Summary

• Global singleton: The event bus lives on globalThis under the symbol @agent-native/core/event-bus.bus, ensuring all modules share the same communication channel.
• Type safety: Zod schemas in registerEvent validate payloads before emission, aborting invalid events and logging warnings.
• Snapshot dispatch: The bus snap-shots listeners before emitting, preventing mid-dispatch modifications from affecting delivery.
• Error isolation: Individual handler failures are caught and logged without breaking the event chain or stopping subsequent listeners.
• Automatic metadata: Every event carries an eventId, timestamp, and optional owner metadata for tracing real-time flows.

Frequently Asked Questions

How does the event bus ensure type safety across components?

The event bus enforces type safety through Zod schemas registered via registerEvent in packages/core/src/event-bus/registry.ts. When emit is called, the bus validates the payload against the stored schema before dispatching. Valid payloads are coerced to the correct types; invalid payloads abort the emission and trigger a warning, preventing type errors from propagating through the real-time communication system.

What happens if an event handler throws an error during real-time processing?

The event bus implements error isolation in packages/core/src/event-bus/bus.ts (lines 101-120). Synchronous exceptions are caught with try/catch and logged without stopping subsequent listeners. Asynchronous rejections are handled with .catch() to prevent unhandled promise rejections. This ensures that a single faulty handler cannot break the entire event chain or disrupt other components' real-time communication.

Is the event bus shared across different processes or servers?

No, the event bus is an in-process singleton that lives on globalThis within a single JavaScript runtime. It facilitates real-time communication between components inside the same process—such as the Vite dev server, Nitro server, or built bundle—but does not cross process boundaries. For multi-process communication, you would need an external message broker or network transport layer.

How do I prevent memory leaks when using long-lived event subscriptions?

Always store the subscription ID returned by subscribe() and call unsubscribe(id) when the component unmounts or the action completes. The unsubscribe function safely removes the handler from the internal EventEmitter and deletes the record from the subscriptions Map, ensuring that inactive handlers are garbage collected and do not accumulate in memory.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}