suaveplan/openspec/specs/events/spec.md

events

Purpose

Typed event-bus substrate for the Suaveplan monorepo — pub/sub with deterministic FIFO delivery, interceptor API, bounded-queue backpressure, idempotent listener wrappers, Zod-validated envelopes, and a passive observability adapter. Supports buffering for late subscribers, wildcard pattern matching, async iteration, concurrency policies (sequential/parallel/allSettled), and AbortSignal/timeout controls on async dispatch.

Requirements

Requirement: Zero-framework runtime dependencies

The @suaveplan/events package's runtime dependencies MUST be limited to @suaveplan/error, tslib, and zod. No peerDependencies; no other @suaveplan/* packages at runtime.

Verified by: tests/dep-graph.test.ts and tests/no-disallowed.test.ts.

Scenario: Dep-graph guard

• WHEN dependencies is read from package.json
• THEN the set equals { @suaveplan/error, tslib, zod }

Scenario: No Node / browser built-ins leak

• WHEN src/ is scanned for node:*, process, Buffer, document, window
• THEN zero matches appear outside an approved defaults file

Requirement: Typed event bus

The package SHALL export an EventBus<TMap> class where TMap extends Record<string, unknown>. on, once, off, emit, and emitAsync are constrained by TMap so payload is inferred from type.

Verified by: src/bus/event-bus.test.ts (typed bus scenarios).

Scenario: Payload type narrows by event name

• WHEN a consumer declares new EventBus<{ login: { user: string } }>() and calls bus.emit("login", x)
• THEN x is typed { user: string } (not unknown)

Scenario: Unknown event name is a type error

• WHEN a consumer calls bus.emit("does-not-exist", ...) on a bus with no such key
• THEN TypeScript rejects at compile time

Requirement: Deterministic FIFO delivery

Listeners registered for a given event SHALL fire in registration order. Across events, synchronous emit delivers fully before returning; async emitAsync awaits every listener in registration order by default.

Verified by: src/bus/event-bus.test.ts (sync + async ordering scenarios).

Scenario: Sync order

• WHEN listeners A, B, C are registered on "x" and emit("x", v) runs
• THEN A runs to completion, then B, then C

Scenario: Async order

• WHEN async listeners A, B are registered on "x" and emitAsync("x", v) runs
• THEN A's returned promise settles before B is invoked

Requirement: Interceptor API

The package SHALL export bus.use(interceptor) registering an interceptor of shape (envelope, next) => Promise<EmitResult>. Interceptors compose in registration order (first-registered outermost). An interceptor MAY omit calling next() to drop the event without invoking listeners.

Verified by: src/interceptor/compose.test.ts and src/bus/event-bus.test.ts (interceptor scenarios).

Scenario: Interceptor can short-circuit

• WHEN an interceptor returns without calling next()
• THEN registered listeners do not fire AND EmitResult.delivered === 0

Scenario: Interceptor throw aborts emit

• WHEN an interceptor throws
• THEN emitAsync rejects with EventsError("EVENTS.INTERCEPTOR.THREW") with cause === the thrown value

Scenario: Interceptor can replace payload

• WHEN an interceptor passes a new envelope to next()
• THEN downstream listeners observe the replaced payload

Requirement: Structured errors

All thrown errors SHALL be EventsError extends SuaveplanError<EventsCode> with codes prefixed EVENTS.. Listener failures SHALL be collected into EmitResult.errors, not propagated, unless an interceptor explicitly re-throws. This applies to both emit (sync) and emitAsync (async) — both paths capture listener throws into EmitResult.errors rather than rejecting the dispatch.

Verified by: src/errors/events-error.test.ts, src/bus/event-bus.test.ts, and src/conformance.spec-conformance.test.ts (sync + async listener-throw scenarios).

Scenario: Sync listener throw is captured

• WHEN a sync listener throws on emit
• THEN emit returns EmitResult with errors[0].code === "EVENTS.LISTENER.THREW" AND errors[0].cause === the thrown value AND delivered reflects successful deliveries only

Scenario: Async listener throw is captured

• WHEN an async listener rejects on emitAsync
• THEN the returned Promise<EmitResult> resolves (does not reject) AND errors[0].code === "EVENTS.LISTENER.THREW" AND errors[0].cause === the rejection value AND delivered reflects successful deliveries only

Requirement: Backpressure via BoundedQueue

The package SHALL export a BoundedQueue<T> class with { capacity, policy } where policy is one of "drop-oldest", "drop-newest", "block", "throw". enqueue(item, signal?) honours AbortSignal.

Verified by: src/queue/bounded-queue.test.ts.

Scenario: drop-oldest evicts head on overflow

• WHEN a 2-capacity queue with "drop-oldest" receives 3 items
• THEN only the latest 2 remain, head was evicted

Scenario: throw policy raises EVENTS.QUEUE.OVERFLOW

• WHEN a 1-capacity queue with "throw" receives a 2nd enqueue
• THEN the 2nd enqueue throws EventsError("EVENTS.QUEUE.OVERFLOW")

Scenario: block waits for capacity

• WHEN a 1-capacity queue with "block" is full AND a 2nd enqueue is pending AND the head is dequeued
• THEN the pending enqueue resolves

Scenario: Aborted enqueue rejects

• WHEN enqueue(x, signal) is pending under "block" AND the signal aborts
• THEN the pending enqueue rejects with EventsError("EVENTS.QUEUE.ABORTED")

Scenario: Invalid capacity rejected

• WHEN a queue is constructed with capacity <= 0 or non-integer capacity
• THEN construction throws EventsError("EVENTS.QUEUE.INVALID_CAPACITY")

Requirement: Idempotent listener wrapper

The package SHALL export idempotent(listener, { keyOf, cache? }) returning a listener that fires at most once per keyOf(payload).

Verified by: src/idempotent/idempotent.test.ts.

Scenario: Duplicate key suppressed

• WHEN the wrapped listener is invoked with two payloads yielding the same key
• THEN the underlying listener fires exactly once

Scenario: Distinct keys fire independently

• WHEN the wrapped listener is invoked with payloads yielding different keys
• THEN the underlying listener fires for each

Requirement: Deterministic id and clock sources

Event envelope id and timestamp SHALL come from injectable sources. The EventBus constructor accepts new EventBus<TMap>({ clock?, idGenerator? }) overrides; there is no createBus factory. The defaults live in two named exports (defaultClock, defaultIdGenerator) and nowhere else in src/.

Verified by: tests/no-disallowed.test.ts and src/conformance.spec-conformance.test.ts (constructor injection scenario).

Scenario: No module-scope non-determinism

• WHEN src/ is scanned for Date.now, crypto.randomUUID, Math.random
• THEN those symbols appear only in clock/default-clock.ts and id/id-generator.ts

Scenario: Injected clock and id generator flow to envelopes

• WHEN a bus is created with new EventBus({ clock: () => 42, idGenerator: () => "X" }) and an event is emitted
• THEN the produced envelope has timestamp === 42 AND id === "X"

Requirement: Passive observability adapter

The package SHALL export withObservability(bus, adapter) returning an unsubscribe handle. The adapter SHALL implement an optional subset of { onEmit, onEmitComplete, onEmitError }. No OpenTelemetry SDK is imported.

Verified by: src/observability/with-observability.test.ts and src/conformance.spec-conformance.test.ts (onEmitError scenario).

Scenario: Adapter hooks receive envelope and result on success

• WHEN an event is emitted through the wrapped bus and the dispatch chain resolves
• THEN adapter.onEmit(envelope) fires once before listeners AND adapter.onEmitComplete(envelope, result) fires once after AND adapter.onEmitError does NOT fire

Scenario: Adapter onEmitError fires when dispatch rejects

• WHEN an interceptor downstream of withObservability throws OR emitAsync times out OR emitAsync is aborted
• THEN adapter.onEmitError(envelope, error) fires once with the rejecting EventsError AND the rejection propagates to the caller AND adapter.onEmitComplete does NOT fire

Requirement: Test coverage at 100%

vitest.config.ts MUST enforce statements: 100, branches: 100, functions: 100, lines: 100. No istanbul ignore pragmas (they do not survive esbuild). Property-based tests cover FIFO order, interceptor composition, queue policies, and idempotency.

Verified by: vitest.config.ts thresholds and the full package test suite under bunx turbo test:coverage --filter=@suaveplan/events.

Scenario: Coverage threshold enforced

• WHEN bunx turbo test:coverage --filter=@suaveplan/events runs
• THEN it exits non-zero if any metric drops below 100

Requirement: emitAsync timeout

emitAsync(type, payload, { timeoutMs }) SHALL reject with EventsError("EVENTS.EMIT.TIMED_OUT") if the dispatch chain (interceptors + listeners) has not settled within timeoutMs milliseconds. Omitting timeoutMs disables the timeout entirely. Settlement before the deadline cancels the timer; the timeout MUST NOT leak a pending handle into the test runner.

Verified by: src/conformance.spec-conformance.test.ts (timeout scenario) and src/bus/event-bus.test.ts.

Scenario: Timeout rejects with EVENTS.EMIT.TIMED_OUT

• WHEN emitAsync("x", v, { timeoutMs: 5 }) is invoked AND a registered listener takes longer than 5ms to settle
• THEN the returned promise rejects with EventsError AND error.code === "EVENTS.EMIT.TIMED_OUT" AND error.context.timeoutMs === 5

Scenario: No timeoutMs means no timer

• WHEN emitAsync("x", v) is invoked with no timeoutMs
• THEN the returned promise settles solely on dispatch outcome AND no setTimeout is scheduled

Requirement: emitAsync abort

emitAsync(type, payload, { signal }) SHALL honour an AbortSignal. If the signal is already aborted at dispatch start OR aborts between listener invocations, dispatch SHALL throw EventsError("EVENTS.EMIT.ABORTED") with cause === signal.reason. Listeners that have already started SHALL run to their natural completion (cooperative cancellation, not preemption).

Verified by: src/conformance.spec-conformance.test.ts (abort scenarios) and src/bus/event-bus.test.ts.

Scenario: Pre-aborted signal rejects emitAsync

• WHEN emitAsync("x", v, { signal }) is invoked with signal.aborted === true and signal.reason === "stop"
• THEN the returned promise rejects with EventsError AND error.code === "EVENTS.EMIT.ABORTED" AND error.cause === "stop"

Scenario: Mid-dispatch abort halts subsequent listeners

• WHEN listeners A, B are registered AND emitAsync("x", v, { signal }) runs AND A aborts the signal during its body
• THEN A runs to completion AND B is NOT invoked AND the returned promise rejects with EventsError("EVENTS.EMIT.ABORTED")

Requirement: Envelope traceId propagation

emit(type, payload, traceId?) and emitAsync(type, payload, { traceId? }) SHALL accept an optional traceId string and surface it on the produced EventEnvelope.traceId. Omitting traceId SHALL leave the envelope's traceId undefined (NOT generated, NOT defaulted). traceId MUST flow through interceptors and observability adapters unchanged.

Verified by: src/conformance.spec-conformance.test.ts (traceId scenarios) and src/envelope/envelope.test.ts.

Scenario: Provided traceId surfaces on envelope

• WHEN emitAsync("x", v, { traceId: "trace-123" }) is invoked AND an interceptor inspects the envelope
• THEN envelope.traceId === "trace-123"

Scenario: Omitted traceId leaves envelope undefined

• WHEN emit("x", v) is invoked with no traceId
• THEN envelope.traceId === undefined AND no fallback id is generated

Requirement: Event buffering for late subscribers

The EventBus constructor SHALL accept bufferSize?: number. When bufferSize > 0, every emitted event SHALL be stored in a rolling buffer per event type. When a new listener is registered via on(), the buffered events for that type SHALL be replayed to the listener synchronously before on() returns. Listener errors during replay SHALL be swallowed.

Verified by: src/bus/event-bus.test.ts (buffering scenarios).

Scenario: Late subscriber receives buffered events

• WHEN emit("x", 1) and emit("x", 2) have already run AND a new listener is registered via on("x", ...)
• THEN the listener is called with 1 then 2 before on() returns

Scenario: Buffer respects size limit

• WHEN bufferSize is 2 AND three events are emitted
• THEN a late subscriber only receives the most recent 2 events

Requirement: Wildcard pattern matching

The EventBus constructor SHALL accept wildcard?: boolean. When wildcard is true, listeners registered on patterns containing * SHALL be invoked for any emit whose type matches the pattern. Exact-match listeners SHALL fire before wildcard listeners.

Verified by: src/bus/event-bus.test.ts (wildcard scenarios).

Scenario: Wildcard subscriber matches multiple events

• WHEN a listener is registered on "plugin:*" AND emit("plugin:ready", 1) runs
• THEN the wildcard listener fires with payload 1

Scenario: Wildcard disabled is default

• WHEN wildcard is omitted or false AND a listener is registered on "plugin:*"
• THEN emit("plugin:ready", 1) does NOT invoke the wildcard listener

Requirement: Async iteration

The EventBus SHALL export iter<K>(type: K, signal?: AbortSignal): AsyncIterable<TMap[K]>. Each emitted event of the given type SHALL be yielded to the for await loop. Calling return() on the iterator or aborting the signal SHALL unsubscribe the listener and end iteration.

Verified by: src/bus/event-bus.test.ts (async iteration scenarios).

Scenario: for await yields emitted payloads

• WHEN bus.iter("x") is consumed in a for await loop AND emit("x", 1) then emit("x", 2) run
• THEN the loop yields 1 then 2

Scenario: Abort ends iteration

• WHEN an AbortSignal passed to iter() is aborted
• THEN the for await loop terminates

Requirement: Listener introspection

The EventBus SHALL export:

• listeners<K>(type: K): Listener<TMap[K]>[] — returns a shallow copy of exact-match listeners
• listenerCount<K>(type: K): number — returns the count of exact-match listeners
• offAll<K>(type?: K): void — removes all listeners for the given type, or all listeners if no type is provided

Verified by: src/bus/event-bus.test.ts (listener management scenarios).

Requirement: prependListener

The EventBus SHALL export prepend<K>(type: K, listener: Listener<TMap[K]>): Unsubscribe. The listener SHALL be inserted at the head of the dispatch queue for that event type, so it fires before listeners added via on().

Verified by: src/bus/event-bus.test.ts (prepend scenario).

Scenario: prepended listener fires first

• WHEN on("x", A) then prepend("x", B) then emit("x", v) run
• THEN B fires before A

Requirement: once off round-trip

once(type, listener) SHALL register the original listener in an internal WeakMap so that off(type, originalListener) removes the once wrapper. The wrapper itself SHALL NOT be required for removal.

Verified by: src/bus/event-bus.test.ts (once + off scenario).

Scenario: off with original listener removes once registration

• WHEN once("x", f) is called AND then off("x", f) is called
• THEN emit("x", v) does NOT invoke f

Requirement: Concurrency control

emitAsync SHALL accept concurrency?: "sequential" | "parallel" | "allSettled". The default SHALL be "sequential".

• sequential — listeners await in registration order (existing behaviour).
• parallel — all listeners are invoked concurrently; the promise resolves when all settle.
• allSettled — all listeners are invoked concurrently; the promise resolves when all settle, collecting every rejection into EmitResult.errors.

Verified by: src/bus/event-bus.test.ts (concurrency scenarios).

Scenario: parallel runs listeners concurrently

• WHEN two slow listeners are registered AND emitAsync is called with concurrency: "parallel"
• THEN both listeners run concurrently, not sequentially

Scenario: allSettled collects all errors

• WHEN two listeners throw AND emitAsync is called with concurrency: "allSettled"
• THEN both errors appear in EmitResult.errors AND delivered === 0

Requirement: maxListeners warning

The EventBus constructor SHALL accept maxListeners?: number (default 10). When the listener count for any event type exceeds this threshold, the bus SHALL emit a warning via the errorSink callback if one is configured.

Verified by: src/bus/event-bus.test.ts (maxListeners scenario).

Scenario: maxListeners exceeded routes to errorSink

• WHEN maxListeners: 2 is configured AND a third listener is added
• THEN errorSink is called with an EventsError whose code is EVENTS.LISTENER.MAX_EXCEEDED

Requirement: errorSink

The EventBus constructor SHALL accept errorSink?: (error: EventsError) => void. Every listener error captured during emit or emitAsync SHALL be routed to errorSink in addition to being placed in EmitResult.errors.

Verified by: src/bus/event-bus.test.ts (errorSink scenarios).

Scenario: errorSink receives sync listener errors

• WHEN a sync listener throws AND errorSink is configured
• THEN errorSink is called with the EventsError

Scenario: errorSink receives async listener errors

• WHEN an async listener rejects AND errorSink is configured
• THEN errorSink is called with the EventsError

Requirement: Disposable

EventBus SHALL implement Symbol.dispose. Calling [Symbol.dispose]() SHALL call clear(). After disposal, subsequent calls to emit or emitAsync SHALL throw EventsError("EVENTS.EMIT.ABORTED").

The Unsubscribe handle returned by on, once, prepend, and use SHALL also implement Symbol.dispose, equivalent to calling the function.

Verified by: src/bus/event-bus.test.ts (disposal scenarios).

Scenario: Bus dispose clears state

• WHEN [Symbol.dispose]() is called on a bus with listeners
• THEN subsequent emit throws AND subsequent emitAsync rejects

Scenario: Unsubscribe handle dispose removes listener

• WHEN const unsub = on("x", f) AND unsub[Symbol.dispose]() is called
• THEN emit("x", v) does NOT invoke f