Event streaming

What event streaming is

Event streaming gives you a persistent, real-time connection to theAuth events via Server-Sent Events (SSE). Whenever your code calls stream.emit() (an agent is revoked, a budget is exceeded, and so on), connected clients receive it, no polling required. Events are also persisted to the database so you can replay anything you missed.

Streaming vs webhooks

Webhooks and SSE are separate mechanisms with separate event sets: the webhook module delivers the events you pass to theauth.webhooks.emit(), and the stream delivers the events you pass to stream.emit(). You can emit the same occurrence to both.

	Webhooks	Event streaming
Transport	HTTP POST to your server	Persistent HTTP connection to your browser or service
Latency	Network latency plus retries	Near-instant
Auth	Signed with an HMAC signature header	Bearer token
Missed events	Retried up to 3 times by default, then dropped	Replay via the `since` timestamp
Best for	Backend integrations, data pipelines	Dashboards, SOC tooling, live monitoring

Use webhooks when you need durable delivery to an external service. Use event streaming when you need a live view, a security dashboard, an admin feed, or a CI script watching for budget.exceeded.

Setup

import { createTheAuth } from '@glinr/theauth';
import { createEventStreamModule } from '@glinr/theauth/auth';

const theauth = await createTheAuth({
  database: { provider: 'sqlite', url: 'theauth.db' },
});

const stream = createEventStreamModule({
  db: theauth.db,
  requireAuth: true,
  validateToken: async (token) => {
    // Return a subscriber ID (userId or agentId) on success, null on failure.
    // Here: a TheAuth-managed session (requires `auth.session` in the config).
    const session = await theauth.auth.session?.validate(token);
    return session?.userId ?? null;
  },
});

// Mount in a Web-standard (Request/Response) handler, for example a Next.js route
export async function GET(request: Request): Promise<Response> {
  return stream.handleRequest(request) ?? new Response('Not found', { status: 404 });
}

// Emit from anywhere in your application
stream.emit({
  id: crypto.randomUUID(),
  type: 'agent.created',
  timestamp: new Date(),
  data: { agentId: agent.id, name: agent.name },
});

handleRequest takes a Web Request and returns a Web Response, or null for any request that is not an SSE request: the method must be GET, the path must end in /events/stream, and the Accept header must include text/event-stream. This makes it safe to call inside a catch-all handler. Node (req, res) handlers such as Express need a small adapter to convert to and from Request and Response.

Connecting from a browser

The browser’s built-in EventSource API handles reconnection automatically.

const source = new EventSource(
  '/api/theauth/events/stream?token=your-bearer-token',
);

// Each message is named by its type and carries { id, type, timestamp, data, agentId?, userId? }
source.addEventListener('agent.created', (event) => {
  const payload = JSON.parse((event as MessageEvent).data);
  console.log('New agent:', payload.data.agentId);
});

source.addEventListener('budget.exceeded', (event) => {
  const payload = JSON.parse((event as MessageEvent).data);
  console.warn('Budget exceeded:', payload.data);
});

source.addEventListener('error', (event) => {
  // Server-sent error events carry JSON; connection errors have no data
  const raw = (event as MessageEvent).data;
  if (typeof raw === 'string') {
    console.error('Stream error:', JSON.parse(raw).code);
  }
});

// Clean up
source.close();

If you pass the token via the Authorization header instead, use the eventsource package or a fetch-based polyfill, since the native EventSource does not support custom headers.

Connecting from Node.js

import { EventSource } from 'eventsource';

const source = new EventSource('https://your-app.com/api/theauth/events/stream', {
  headers: {
    Authorization: 'Bearer your-bearer-token',
  },
});

// Messages are named events, so onmessage never fires. Listen per type.
source.addEventListener('agent.revoked', (event) => {
  const payload = JSON.parse(event.data);
  console.log('Event:', payload.type, payload.data);
});

source.onerror = () => {
  console.error('Connection lost, reconnecting...');
};

Event types

The stream module is standalone. The theAuth core does not call stream.emit() for you, so none of these events fire automatically: you emit the ones you need from your own code, hooks, or webhook handlers. anomaly.detected is only a reserved type name, since theAuth has no anomaly detector.

The type column below describes the intended meaning of each type.

Type	When it fires
`audit`	Any agent action logged to the audit trail
`agent.created`	A new agent identity is registered
`agent.revoked`	An agent is permanently revoked
`agent.rotated`	An agent token is rotated
`auth.signin`	A user signs in
`auth.signout`	A user signs out
`auth.failed`	A sign-in attempt fails
`delegation.created`	A new delegation chain is created
`delegation.revoked`	A delegation is revoked
`budget.exceeded`	An agent exceeds its cost budget
`anomaly.detected`	Reserved name. theAuth has no anomaly detector, so this only appears if your own code emits it
`cost.recorded`	A cost event is attributed to an agent

Filtering events

Pass a types query parameter with a comma-separated list to receive only the events you care about.

GET /api/theauth/events/stream?types=agent.revoked,budget.exceeded

// From a browser
const source = new EventSource(
  '/api/theauth/events/stream?token=tok&types=agent.created,agent.revoked',
);

Unknown type names in types are ignored; if none are valid, the connection receives all types. You can also restrict the types at the module level, which filters live delivery.

const stream = createEventStreamModule({
  db: theauth.db,
  requireAuth: true,
  validateToken,
  eventTypes: ['audit', 'budget.exceeded'], // only these types are delivered live
});

Replay and cursor

Events passed to emit() are persisted in the theauth_stream_events table (a failed write is ignored). If a client disconnects and reconnects, pass since to receive everything it missed, oldest first, after the connected event. since must be an ISO 8601 timestamp with an offset or Z; anything else is rejected with a 400.

GET /api/theauth/events/stream?since=2026-01-15T10:00:00Z

The module also reads the Last-Event-ID header as a fallback cursor, but it parses the value as a date. Each SSE message uses your event id as its id: field, so the header only works for replay if your event IDs are timestamps (for example ISO strings). With UUID IDs, as in the examples here, an automatic browser reconnect replays nothing, so track the last timestamp yourself and pass since. Replay applies the connection’s types filter, or the module-level eventTypes if the client sent none. A client that passes its own types can replay types outside the module-level list, and the programmatic replay() method does not apply the module-level list at all.

// Programmatic replay without an open connection
const result = await stream.replay(new Date('2026-01-15T10:00:00Z'), ['audit', 'auth.failed']);
if (result.success) {
  for (const event of result.data) {
    console.log(event.type, event.data);
  }
}

replay returns up to 1000 events in descending order (newest first). Apply your own pagination on top if you need to page through large windows.

Auth requirements

By default requireAuth: true. Every connection must present a Bearer token, either in the Authorization header or as the token query parameter.

Authorization: Bearer <token>
GET /api/theauth/events/stream?token=<token>

When the token is missing or validateToken returns null, the stream sends a single error event and closes (the HTTP status is still 200).

{ "code": "UNAUTHORIZED", "message": "Invalid token" }

If you set requireAuth: true but do not provide validateToken, any non-empty token is accepted. Always supply validateToken in production.

To disable auth for local development or internal-only deployments:

const stream = createEventStreamModule({
  db: theauth.db,
  requireAuth: false,
});

Never disable auth in production. The stream exposes audit events and agent lifecycle data.

Configuration

interface EventStreamConfig {
  db: Database;

  // Maximum concurrent connections. Returns 503 when exceeded (default: 100).
  maxConnections?: number;

  // Interval between heartbeat comments in ms (default: 30000)
  heartbeatIntervalMs?: number;

  // Module-level event type allow-list. Clients cannot exceed this (default: all).
  eventTypes?: EventType[];

  // Require a Bearer token to connect (default: true)
  requireAuth?: boolean;

  // Validate the token and return a subscriber ID (userId or agentId), or null.
  // If omitted while requireAuth is true, any non-empty token is accepted.
  validateToken?: (token: string) => Promise<string | null>;
}

Connection limits

The stream rejects connections beyond maxConnections with a 503 Too many connections response. Size this based on your deployment: a single process can comfortably handle hundreds of concurrent SSE connections; above that, consider a pub/sub layer (Redis, NATS) in front of the module.

Heartbeat

The server sends : heartbeat comments on the configured interval (default 30 seconds) to keep load balancers and proxies from closing idle connections. No action is required on the client side, EventSource ignores comment lines.

Emitting events from plugins

Any part of your application can emit to the stream.

// After revoking an agent
await theauth.agent.revoke(agentId);

stream.emit({
  id: crypto.randomUUID(),
  type: 'agent.revoked',
  timestamp: new Date(),
  data: { agentId, reason: 'manual-revocation' },
  agentId,
  userId: currentUser.id,
});

Integrate with the webhooks module to fire both a webhook and a stream event from the same action.

Module API

interface EventStreamModule {
  // Emit an event to all connected clients and persist for replay
  emit(event: StreamEvent): void;

  // Handle an SSE connection request. Returns a Response or null.
  handleRequest(request: Request): Response | null;

  // Number of currently active connections
  getConnectionCount(): number;

  // Replay persisted events since a date, optionally filtered by type.
  // Up to 1000 events, newest first.
  replay(since: Date, types?: EventType[]): Promise<Result<StreamEvent[]>>;

  // Close all connections and stop the heartbeat timer
  close(): void;
}

Webhooks

Durable HTTP delivery for the same events to external services.

Audit trail

The audit log. Replay reads from the separate stream events table, not from audit entries.

Hooks

Call stream.emit() from lifecycle hooks such as onViolation and onAgentRevoke.

Dashboard

Visual monitoring of agents and audit entries.

Compliance and security

Lifecycle

Identity primitives

Standards

Event streaming

What event streaming is

Streaming vs webhooks

Setup

Connecting from a browser

Connecting from Node.js

Event types

Filtering events

Replay and cursor

Auth requirements

Configuration

Connection limits

Heartbeat

Emitting events from plugins

Module API

Webhooks

Audit trail

Hooks

Dashboard

​What event streaming is

​Streaming vs webhooks

​Setup

​Connecting from a browser

​Connecting from Node.js

​Event types

​Filtering events

​Replay and cursor

​Auth requirements

​Configuration

​Connection limits

​Heartbeat

​Emitting events from plugins

​Module API

​Related

Webhooks

Audit trail

Hooks

Dashboard

What event streaming is

Streaming vs webhooks

Setup

Connecting from a browser

Connecting from Node.js

Event types

Filtering events

Replay and cursor

Auth requirements

Configuration

Connection limits

Heartbeat

Emitting events from plugins

Module API

Related