# EventHub

A bounded tail of recent events that wakes waiters and serves Server-Sent Events.

`EventHub` keeps the most recent contiguous events from `published` in memory, bounded by an event count and by total encoded bytes. Readers inside the tail read from memory, and readers behind it page from the log. `stream` serves a finite or open Server-Sent Events response from pages that the host reads.

```ts title="events.ts"
import { AgentRuntime, EventHub, type SqlHost } from 'nanocode-wasm'

declare const module: WebAssembly.Module
declare const sql: SqlHost

// 1. Keep at most 128 events and 4 MiB in the tail
const hub = new EventHub(
  () => 128,
  () => 4 * 1024 * 1024,
)
const runtime = await AgentRuntime.instantiate(module, sql, {
  published: (events, bytes) => hub.publish(events, bytes),
})

// 2. Serve the log from the tail, or from storage for readers behind it
export function events(after: number): Response {
  return hub.stream(
    after,
    15_000,
    async (cursor) =>
      hub.recent(cursor) ?? runtime.page('owner', cursor, 128),
  )
}
```

## Signature

```ts
class EventHub {
  constructor(retained: () => number, retainedBytes: () => number)
  get version(): number
  notify(): void
  publish(events: LoggedEvent[], bytes: number[]): void
  recent(after: number, requestId?: string): LogPage | undefined
  wait(milliseconds: number, version?: number): Promise<void>
  stream(
    after: number,
    interval: number,
    read: (after: number) => Promise<LogPage>,
    finished?: (
      event: LoggedEvent & { [key: string]: unknown },
    ) => boolean,
  ): Response
}
```

## Parameters

### retained

* **Type:** `() => number`

Returns the largest number of events that the tail keeps. The hub reads it on every `publish`, so the host can change the bound at run time.

```ts
import { EventHub } from 'nanocode-wasm'

// Reads the bound from configuration on every publish
declare const config: { pageSize: number; tailBytes: number }
const hub = new EventHub(() => config.pageSize, () => config.tailBytes)
```

### retainedBytes

* **Type:** `() => number`

Returns the largest total encoded size of the tail in bytes. The hub drops the oldest events first. One event above this bound leaves the tail empty.

```ts
import { EventHub } from 'nanocode-wasm'

// Bounds the tail by memory as well as count
const hub = new EventHub(() => 128, () => 4 * 1024 * 1024)
```

## Bounds

The count alone does not bound memory, because the size of an event varies. On 2026-10-20 in Node 22.23, `scripts/measure-events.mjs` published 128 events of each size into a hub bounded by 128 events and 4 MiB:

| Event size | Count-only tail   | Tail with the 4 MiB bound    |
| ---------- | ----------------- | ---------------------------- |
| 300 B      | Not measured      | 128 events in 129,288 bytes  |
| 4 KiB      | Not measured      | 128 events in 583,032 bytes  |
| 64 KiB     | 8,437,864 bytes   | 63 events in 4,152,544 bytes |
| 1 MiB      | 134,266,928 bytes | 3 events in 3,146,448 bytes  |

The count-only tail of 1 MiB events retained more than an isolate's 128 MB. Parsed event text used about one byte of heap per encoded byte.

## Methods

### publish

```ts
// Takes the same arguments as HostOptions.published
publish(events: LoggedEvent[], bytes: number[]): void
```

Appends a batch to the tail and wakes waiters. `bytes[i]` is the encoded size of `events[i]`. A batch that does not follow the tail's last `seq` clears the tail, so the tail stays contiguous.

```ts
import { EventHub, type LoggedEvent } from 'nanocode-wasm'

declare const hub: EventHub
declare const events: LoggedEvent[]
declare const bytes: number[]

// Forwards one committed batch
hub.publish(events, bytes)
```

### recent

```ts
// undefined when the reader is behind the tail
recent(after: number, requestId?: string): LogPage | undefined
```

Returns the tail's events after `after`, optionally for one turn, as a `LogPage` with `has_more: false`. If the tail is empty or does not reach back to `after`, it returns `undefined`, and the reader pages from the log with [`page`](/docs/api/wasm/page).

```ts
import type { AgentRuntime, EventHub } from 'nanocode-wasm'

declare const hub: EventHub
declare const runtime: AgentRuntime

// Reads from memory first, then from storage
const page = hub.recent(41) ?? (await runtime.page('owner', 41, 128))
```

### wait

```ts
// Resolves on the next publish or notify, or after the timeout
wait(milliseconds: number, version?: number): Promise<void>
```

Waits up to `milliseconds` for the next `publish` or `notify`. If `version` differs from the current `version`, it resolves at once, so a reader that read `version` before a check misses no change.

```ts
import type { EventHub } from 'nanocode-wasm'

declare const hub: EventHub
declare function check(): boolean

// Reads the version before the check to avoid a lost wake
const version = hub.version
if (!check()) await hub.wait(1_000, version)
```

### notify

```ts
// Increments version and wakes every waiter and stream
notify(): void
```

Wakes waiters without a new batch, for example after a turn's status changes.

```ts
import type { EventHub } from 'nanocode-wasm'

declare const hub: EventHub

// Wakes streams so they read the next page now
hub.notify()
```

### stream

```ts
// A text/event-stream Response
stream(
  after: number,
  interval: number,
  read: (after: number) => Promise<LogPage>,
  finished?: (
    event: LoggedEvent & { [key: string]: unknown },
  ) => boolean,
): Response
```

Serves events after `after` as Server-Sent Events. Each event is sent as `id: <seq>` and `data:` with the event JSON plus `streamed_at`. If no event arrives within `interval` milliseconds, the stream sends a `: keepalive` comment. If `finished` matches an event in a page, the stream closes after that page.

```ts
import type { AgentRuntime, EventHub } from 'nanocode-wasm'

declare const hub: EventHub
declare const runtime: AgentRuntime

// Streams one turn and closes at its result event
const response = hub.stream(
  0,
  15_000,
  (cursor) => runtime.page('owner', cursor, 128, 'turn-1'),
  (entry) => (entry.event as { type?: string }).type === 'result',
)
```

The response has `cache-control: no-store`. If `read` rejects, the stream errors with `event stream unavailable`. Cancelling the response body stops the stream.

## Errors

`publish`, `recent`, `notify`, and `wait` do not throw. `stream` returns its response at once and reports a failed `read` as a stream error.

```ts
import type { EventHub } from 'nanocode-wasm'

declare const hub: EventHub

// The response is returned even if the first read fails later
const response = hub.stream(0, 15_000, async () => {
  throw new Error('storage unavailable')
})
```