@zakkster/lite-stream

Zero-GC bridge between async iterators and @zakkster/lite-signal. The multi-shot dual of lite-await's fromPromise: project an async source of N values (paginated APIs, SSE streams, network frame queues, pubsub topics) into a signal-shaped reactive surface, with bounded buffering and structural cleanup on every termination path.

Why this exists

The hand-rolled version of "drive a signal from an async iterator" is a five-line for-await loop that leaks the iterator forever the moment the consumer navigates away, the AbortSignal aborts, or the parent component unmounts. Every place in the @zakkster/lite-* ecosystem that handles a multi-shot async source -- helix pagination, EBS server-sent events, pubsub topic subscriptions, rollback input-frame queues -- would otherwise write that broken loop. This library is the one correct implementation.

+-------------------------+        +-----------------+        +-----------------+
|  AsyncIterable<T>       |        |  lite-stream    |        |  Signal<State>  |
|  (helix pagination,     | -----> |  fromAsync-     | -----> |  (your effect() |
|   EBS SSE, pubsub,      |        |  Iterable       |        |   reads this)   |
|   rollback frames)      |        +-----------------+        +-----------------+
+-------------------------+                |
                                           |  three structural cleanup paths:
                                           |    (a) iterator done    -> done: true
                                           |    (b) iterator throws  -> error: ...
                                           |    (c) AbortSignal abort -> iter.return()
                                           v
                                  no leaks. no listeners outliving the stream.

Install

npm install @zakkster/lite-stream @zakkster/lite-signal

@zakkster/lite-signal is a peer dependency (^1.2.0). No other runtime deps.

Quick start

import { fromAsyncIterable } from "@zakkster/lite-stream";
import { effect } from "@zakkster/lite-signal";

const ctrl = new AbortController();

const live = fromAsyncIterable(networkFrames(), {
    signal: ctrl.signal,
    onDone:  () => console.log("frame stream ended"),
    onError: (e) => console.error("frame stream error", e)
});

effect(() => {
    const s = live();
    if (s.error)        renderError(s.error);
    else if (s.done)    renderEndCard();
    else if (s.value)   renderFrame(s.value);
});

// Later -- consumer navigates away:
ctrl.abort();   // iterator.return() called; signal settles; no leaks

Table of contents

• The cleanup termination triplet
• API reference
  • fromAsyncIterable
  • pipeToSignal
  • createSignalWriter
  • toAsyncIterable
• Modes: latest vs buffer
• Zero-GC hot paths
• Integration recipes
• Edge cases pinned down
• Benchmarks
• Testing strategy
• What this is not
• Ecosystem

The cleanup termination triplet

A stream-to-signal bridge ends on exactly three paths. lite-stream handles all three structurally:

1. Iterator natural completion -- the source yields { done: true }. The signal state settles with done: true. onDone fires once. No further state mutations occur.

2. Iterator throws -- the source rejects a pull or throws synchronously from next(). The signal state settles with error: <thrown> and done: true. onError fires once with the thrown value. The last successfully yielded value is preserved.

3. AbortSignal aborts -- the caller-provided opts.signal enters the aborted state. iter.return() is called best-effort to give the generator a chance to clean up. The signal state settles with error: <AbortError> and done: true. onError fires.

The abort listener is registered with addEventListener("abort", ...) and always removed on any of the three paths -- no AbortSignal accumulates dangling listeners across stream lifecycles.

Disposing the result signal does NOT stop the pump: the iterator keeps pulling. AbortSignal or natural completion are the only stop mechanisms. iterator.return() is never called and no callback fires, because lite-signal's set() after dispose() is a silent no-op -- the pump never learns of the disposal. Abort opts.signal to stop early. See ROADMAP LS-01.

API reference

fromAsyncIterable

fromAsyncIterable<T>(
    source: AsyncIterable<T> | AsyncIterator<T> | Iterable<T>,
    opts?: {
        mode?:      "latest" | "buffer",     // default "latest"
        maxBuffer?: number,                   // REQUIRED if mode = "buffer"
        initial?:   T,                        // initial signal value
        signal?:    AbortSignal,              // abort to stop
        onError?:   (err: unknown) => void,
        onDone?:    () => void
    }
): Signal<State<T>>

Drive a signal from an async iterator. Returns a Signal whose value is a tagged state object reflecting the iterator's lifecycle.

State shape varies by mode:

// "latest" mode
{ value: T | undefined, count: number, done: boolean, error: unknown }

// "buffer" mode
{ values: T[], count: number, droppedCount: number, done: boolean, error: unknown }

count increments on every yielded value (regardless of whether it survives buffering). done flips to true on any of the three termination paths. error is undefined on natural completion, populated otherwise.

pipeToSignal

pipeToSignal<T>(
    source: AsyncIterable<T> | AsyncIterator<T>,
    target: Signal<T>,
    opts?: {
        signal?:    AbortSignal,
        mode?:      "latest" | "buffer",     // default "latest"
        maxBuffer?: number,                   // REQUIRED if mode = "buffer"
        transform?: (value: T) => T,          // per-value map, runs first
        onValue?:   (value: T) => void,       // tap: fires once BEFORE set
        onAbort?:   (reason: unknown) => void, // aborts route here, not onError
        onError?:   (err: unknown) => void,
        onDone?:    () => void
    }
): (() => void) & { readonly droppedCount: number, readonly overflowCount: number }

Lower-level companion: pump an existing writable signal from an async iterator. In the default "latest" mode the signal's value is replaced directly with each yielded value (no { value, count, ... } wrapper). Returns an idempotent stop function that also carries droppedCount / overflowCount getters.

Use pipeToSignal when:

• You already have a signal you want to drive
• You don't need the lifecycle metadata wrapper
• You want a stop fn instead of an AbortController for cleanup

Enrichment (1.3.0, all additive -- the un-optioned path is byte-identical to 1.1.0):

• mode: "buffer" (with required maxBuffer) drives the target with a bounded newest-last snapshot array instead of the raw value; overflow drops the oldest and increments stop.droppedCount. mode: "latest" + maxBuffer throws TypeError; missing/invalid maxBuffer in buffer mode throws RangeError.
• onValue(v) is a per-value tap that fires once BEFORE each set (after transform). A throwing onValue routes through the same stop + onError leg as a throwing transform.
• onAbort(reason), when present, receives aborts INSTEAD of onError, so a consumer aborting on purpose (detach / restart) no longer filters signal.aborted out of its error handler. See the abort-vocabulary contract in llms.txt and decisions/0001-abort-vocabulary.md.

Does NOT dispose the target signal. The caller owns its lifetime. Disposing the target signal does NOT stop the pump: call the returned stop fn or abort opts.signal.

// Enriched buffer-mode pipe -- replaces a hand-rolled ring + status tap +
// abort filter (the StreamQuery shape) with native options.
const stop = pipeToSignal(topicSource, entrySig, {
    signal:    ctrl.signal,
    mode:      "buffer",
    maxBuffer: 64,                               // bounded, drop-oldest
    onValue:   (v) => { if (first) status.set("streaming"); first = false; },
    onAbort:   () => {},                         // intentional abort, not an error
    onError:   (e) => status.set("error"),       // real failures only
    onDone:    () => status.set("done")
});
// stop.droppedCount tells you how many values overflowed the window.

createSignalWriter

createSignalWriter<T>(
    target: Signal<T>,
    opts?: {
        mode?:      "latest" | "buffer",     // default "latest"
        maxBuffer?: number                    // REQUIRED if mode = "buffer"
    }
): {
    push(v: T): void,
    end(): void,
    error(e: unknown): void,
    readonly droppedCount: number
}

The enqueue half of pipeToSignal without the pull pump. Where pipeToSignal OWNS an async iterator and pulls, createSignalWriter is PUSHED: values that arrive by callback enter target with the SAME mode / maxBuffer / droppedCount discipline as pipeToSignal's buffer branch (both drive the target through one shared window implementation -- no third copy of the drop-oldest ring). Added in 1.4.0.

The returned handle is a plain, NON-callable object with exactly four own names. The contract is additive-only: four names now, additive properties only forever.

• push(v) in "latest" mode is one signal write (target.set(v)) -- zero allocation, droppedCount stays 0. In "buffer" mode each push(v) publishes a fresh newest-last snapshot array (exactly one array allocation per push), dropping the oldest on overflow and incrementing droppedCount. Elements are held by reference -- never copied, compared, or serialized. push returns undefined (the window discipline never backpressures).
• end() and error(e) are idempotent terminals: the first wins, later calls in any order are no-ops. NO terminal ever writes the target (an empty writer + end() leaves target unwritten). After a terminal, push is ignored and droppedCount FREEZES at its terminal value. e is unobservable in 1.4.0.
• droppedCount is the buffer-mode overflow count (0 in latest mode), read live before a terminal and the frozen value after.

Option / error matrix (validated synchronously at construction, before any state exists):

| Condition | Result | | --- | --- | | target null / non-object / no .set | TypeError | | unknown mode | TypeError | | maxBuffer with mode: "latest" | TypeError | | mode: "buffer" with missing / non-integer / < 1 / NaN / Infinity maxBuffer | RangeError |

import { createSignalWriter } from "@zakkster/lite-stream";
import { signal } from "@zakkster/lite-signal";

const frame = signal(null);
const writer = createSignalWriter(frame, { mode: "latest" });

// A callback source -- no async iterator to pull:
channel.onmessage = (msg) => writer.push(msg.data);
channel.onclose   = () => writer.end();

toAsyncIterable

toAsyncIterable<T>(
    sig: Signal<T> | Computed<T>,
    opts?: {
        signal?:      AbortSignal,
        emitInitial?: boolean,               // default true
        maxBuffer?:   number,                // default 1024 (buffer mode)

        // Added in 1.1
        mode?:        "latest" | "buffer",   // default "buffer"
        filter?:      (v: T) => unknown,
        timeout?:     number
    }
): AsyncIterable<T> & {
    readonly droppedCount:  number,
    readonly overflowCount: number,          // 1.1 alias for droppedCount
    [Symbol.asyncDispose]?: () => Promise<IteratorResult<T, undefined>>
}

The reverse direction -- yield signal changes as an async iterable. Useful when you have signal-driven state and want to pipe its changes into a WebSocket, log sink, replay tool, or any consumer that wants for-await semantics.

Two backpressure modes:

• "buffer" (default; matches 1.0.0) -- FIFO ring buffer of maxBuffer size. On overflow the OLDEST value is dropped and droppedCount (and overflowCount) increments. Use when every value matters and must be seen in order.
• "latest" (1.1) -- single mutable slot. Producer overwrites; consumer reads and clears. Every overwrite bumps overflowCount. Use for reactive-state consumers ("show the current frame / latest cursor / live price") where intermediate values can be safely discarded.

1.1 additions worth calling out:

• filter -- skip values for which the predicate returns falsy. A throwing filter rejects the current pending next() and terminates the iterable; the throw does NOT surface at the signal writer's .set() call site.
• timeout -- overall deadline in ms; on elapse the pending next() rejects with TimeoutError (a new named export in 1.1). Subsequent calls return {done: true}.
• Symbol.asyncDispose -- delegates to iter.return() on Node 20+; enables await using iter = toAsyncIterable(sig, { mode: "latest" }).
• Multi-waiter queue -- concurrent .next() calls (e.g. from Promise.all([iter.next(), iter.next()])) now resolve in FIFO order. In 1.0.0 the second call silently overwrote the first resolver; the 1.1 fix is invisible to correct callers.

The iterator naturally completes when opts.signal aborts; consumer-side break triggers iterator.return(). mode: "latest" combined with maxBuffer is rejected with TypeError at construction -- a single-slot mode plus a buffer size is a user error worth surfacing loudly.

Treat this as the secondary API for classic "consume signal changes" patterns; for the streaming pipeline case (paginated APIs, SSE, pubsub) the forward direction is what most consumers want.

VERSION

VERSION: string

A string that always equals the installed package.json version. Consumers do NOT re-export it (suite ruling); the surface-guard asserts the equality on every test run.

Modes: latest vs buffer

Pick "latest" when you only care about the most recent value:

• Live cursor positions
• Current pubsub message
• Latest network frame
• Current SSE event
• Reactive state bridged into @zakkster/lite-query's streamQuery (matches its mode: "latest" on the forward direction)

Pick "buffer" when every value matters and you want them in order:

• Helix paginated results (each page must be processed)
• EBS SSE event log (must not miss an event)
• Replay queue (every frame applies in order)

Availability:

• fromAsyncIterable has both modes from 1.0.0.
• toAsyncIterable has both modes from 1.1 (was "buffer"-only in 1.0.0; the option was added as mode: "latest" while preserving default behavior).

"buffer" mode requires maxBuffer on fromAsyncIterable. There is no default there, and missing the option throws a RangeError with this message:

lite-stream: "buffer" mode requires opts.maxBuffer to be a positive integer. Unbounded buffering is a memory bug pretending to be a feature; pick a deliberate ceiling.

toAsyncIterable's "buffer" mode defaults maxBuffer to 1024 (unchanged in 1.1). On overflow the OLDEST value is dropped and droppedCount increments. Watch droppedCount in your effect to know when your consumer is falling behind.

streamQuery integration (matching vocabulary on both directions):

import { toAsyncIterable } from "@zakkster/lite-stream";
import { streamQuery } from "@zakkster/lite-query/stream";

streamQuery(qc, {
    key: ["price", ticker],
    stream: ({ signal }) => toAsyncIterable(priceSig, {
        signal,
        mode: "latest"                     // matches streamQuery's mode
    }),
    mode: "latest"
});

The signal from the stream context wires straight through: streamQuery aborts on detach, toAsyncIterable catches the abort and calls return(). Zero glue code.

Zero-GC hot paths

lite-stream's hot paths -- per-yield state allocation and per-pull abort checks -- minimize per-op allocation. Current numbers with provenance live in llms.txt (Performance); re-run npm run bench on your target.

Per-yield, "latest" mode allocates exactly one wrapper state object. The underlying signal node is reused via lite-signal's pool. "buffer" mode adds one snapshot array per yield (so consumer effects see a stable non-mutated array view at each tick) -- the ring beneath it is a fixed pre-allocated array reused across all yields.

The only per-op allocations beyond intrinsic iterator cost:

• One wrapper state object per yield ("latest") or wrapper + snapshot array ("buffer")
• One AbortListener function and registration on the AbortSignal (once per stream lifetime, not per yield)
• The Promise chain from iter.next().then(...) itself

V8's escape analysis covers the wrapper object closures in most JIT modes.

createSignalWriter carries no pump, no iterator, and no abort listener, so its push path is the tightest in the package. A "latest" push is one signal write with zero steady-state allocation; a "buffer" push allocates exactly one fresh snapshot array (drop-oldest into a pre-allocated ring, then a newest-last copy). Gated per-op numbers from the torture s5 allocation lane (maxMajor 0, node --expose-gc test/torture.mjs):

| Hot path | Steady-state allocation | Measured | Budget | | --- | --- | --- | --- | | createSignalWriter push, "latest" | 0 B (one target.set(v)) | 0.58 B/op | 1 B/op | | createSignalWriter push, "buffer" (maxBuffer 64) | 1 snapshot array | 1.53 B/op | 32 B/op | | pipeToSignal steady pump ("latest") | 0 B steady-state | 1.38 B/op | 32 B/op | | pipeToSignal buffer pump (maxBuffer 64) | 1 snapshot array | 0.68 B/op | 32 B/op |

The writer's "latest" push is proven under 1.0 B/op strictly over 100k pushes with zero major GCs (torture s7 corpus case 11).

Integration recipes

lite-twitch-helix paginated endpoint

import { fromAsyncIterable } from "@zakkster/lite-stream";
import { effect } from "@zakkster/lite-signal";

async function* getFollowers(channelId, signal) {
    let cursor = "";
    while (true) {
        const res = await fetch(
            "https://api.twitch.tv/helix/channels/followers"
                + "?broadcaster_id=" + channelId
                + (cursor ? "&after=" + cursor : ""),
            { signal, headers: { Authorization: bearer, "Client-Id": clientId } }
        );
        const { data, pagination } = await res.json();
        yield data;                                  // yield one page at a time
        cursor = pagination?.cursor;
        if (!cursor) return;
    }
}

const ctrl = new AbortController();
const pages = fromAsyncIterable(getFollowers(channelId, ctrl.signal), {
    mode: "buffer",
    maxBuffer: 100,                                  // 100 pages = 10K followers max
    signal: ctrl.signal
});

effect(() => {
    const s = pages();
    if (s.droppedCount > 0) console.warn("dropped pages:", s.droppedCount);
    renderFollowerList(s.values.flat());
    if (s.done && !s.error) renderDoneCard(s.count);
});

lite-twitch-ebs server-sent events

import { fromAsyncIterable } from "@zakkster/lite-stream";

async function* sseEvents(url, signal) {
    const res = await fetch(url, { signal });
    const reader = res.body.getReader();
    const decoder = new TextDecoder();
    let buffer = "";
    while (true) {
        const { done, value } = await reader.read();
        if (done) return;
        buffer += decoder.decode(value, { stream: true });
        for (const line of buffer.split("\n\n").slice(0, -1)) {
            yield parseSSE(line);
        }
        buffer = buffer.split("\n\n").slice(-1)[0];
    }
}

const ctrl = new AbortController();
const events = fromAsyncIterable(sseEvents("/ebs/events", ctrl.signal), {
    mode: "latest",                                  // only newest matters
    signal: ctrl.signal
});

lite-rollback input-frame queue

import { fromAsyncIterable } from "@zakkster/lite-stream";

async function* gamepadFrames(signal) {
    while (!signal.aborted) {
        await nextAnimationFrame();
        yield sampleGamepad();
    }
}

const frames = fromAsyncIterable(gamepadFrames(ctrl.signal), {
    mode: "buffer",
    maxBuffer: 60,                                   // 1 second at 60fps
    signal: ctrl.signal
});

effect(() => {
    const s = frames();
    for (const frame of s.values) applyInputFrame(frame);
    if (s.droppedCount) console.warn("rollback drops:", s.droppedCount);
});

lite-twitch-pubsub topic subscription

import { pipeToSignal } from "@zakkster/lite-stream";
import { signal } from "@zakkster/lite-signal";

const subs = signal(null);

const ctrl = new AbortController();
const stop = pipeToSignal(subscribeToTopic("channel-subscribe-events.v1"), subs, {
    signal:    ctrl.signal,
    transform: (raw) => JSON.parse(raw.data),
    onError:   (e) => console.error("pubsub error", e)
});

// Later: ctrl.abort() OR stop() -- either ends the pipe.

Callback-sourced frames: a channel feeding a writer

Some sources have no async iterator to pull -- they PUSH: BroadcastChannel messages over lite-query's existing crossTab channel; lite-channel supplies only the caller-wired isLeader oracle. A follower's frames arrive by callback, so createSignalWriter (not pipeToSignal) is the fit -- the writer needs zero channel / epoch awareness.

import { createSignalWriter } from "@zakkster/lite-stream";
import { signal } from "@zakkster/lite-signal";

const events = signal([]);
const writer = createSignalWriter(events, { mode: "buffer", maxBuffer: 64 });

const channel = new BroadcastChannel("query-crosstab");
channel.onmessage = (msg) => {
    // the caller's OR-4 epoch/clientId/seq dedup gate runs BEFORE this line;
    // by here the frame is already accepted -- just enqueue it.
    if (!isLeader()) writer.push(msg.data.value);
};
channel.onmessageerror = (e) => writer.error(e);

// Teardown: end the writer; droppedCount freezes, internal refs are nulled.
function detach() {
    channel.close();
    writer.end();
}

// events() is a fresh newest-last snapshot array per push; overflow drops the
// oldest and bumps writer.droppedCount.

Reverse direction: pipe signal changes into a WebSocket

import { toAsyncIterable } from "@zakkster/lite-stream";

const ctrl = new AbortController();
const ws = new WebSocket("wss://logs.example.com/ingest");

(async () => {
    for await (const event of toAsyncIterable(loggerSig, { signal: ctrl.signal })) {
        if (ws.readyState === WebSocket.OPEN) {
            ws.send(JSON.stringify(event));
        }
    }
})();

Edge cases pinned down

• Pre-aborted AbortSignal: the iterator is never started. The signal settles synchronously with an AbortError. No pulls.
• Synchronous throw from iter.next(): caught and surfaced via the same error path as async rejections.
• Iterator yields null/undefined: passed through as-is. lite-stream doesn't interpret values; only { done: true } terminates.
• Iterator returns a non-object from next(): treated as a protocol error; settles with a TypeError.
• Sync Iterable<T>: accepted via Symbol.iterator. Pulls happen microtask-asynchronously even for sync sources, so caller's effect() can wire up before the first yield.
• Consumer disposes the signal mid-stream: the pump keeps pulling. iterator.return() is not called and no callback fires; abort opts.signal to stop.
• Buffer mode snapshot array: a fresh array per yield. Past observers hold stable, non-mutated references.
• Disposing the signal multiple times: lite-signal's dispose is idempotent; lite-stream tolerates concurrent abort + dispose without double-firing callbacks.

Benchmarks

Run via npm run bench (requires --expose-gc). Current numbers with provenance live in llms.txt (Performance); re-run npm run bench on your target.

Testing strategy

Tier 1 -- behavior (unit tests, fast)

130 tests across test/01-* through test/11-*:

• 01-from-async-iterable-latest.test.mjs -- state shape, lifecycle, Iterable acceptance variants, pre-aborted, subscriber observability
• 02-from-async-iterable-buffer.test.mjs -- ring buffer correctness, drop-oldest, snapshot freshness, maxBuffer validation, helix shape
• 03-cleanup-triplet.test.mjs -- all three termination paths, iterator return() invocation, abort reason propagation, 200-cycle leak smoke test
• 04-pipe-to-signal.test.mjs -- target write, stop fn idempotency, transform, abort behavior, target-not-disposed invariant
• 05-to-async-iterable.test.mjs -- 1.0.0 behavior: emitInitial, ordering, overflow, consumer break triggers return(), pre-aborted path
• 07-to-async-iterable-rich.test.mjs (1.1) -- mode:"latest" behavior, filter (including throwing-filter no-writer-surface discipline), timeout
  • TimeoutError, Symbol.asyncDispose (Node 20+ guarded), multi-waiter queue, overflowCount alias, 4K structural cleanup cycles
• 08-dispose-behavior.test.mjs -- pins the TRUE dispose semantics (LS-01: disposing a result/target signal does NOT stop the pump; LS-04: a pending next() on a disposed source never settles, timeout is the only escape hatch) plus the LS-13 pins (a throwing done/value result getter routes to the error state, onError once, no unhandledRejection)
• 09-guards.test.mjs -- drift guards: ascii-guard (every shipped + test/bench/demo byte is printable ASCII or LF) and surface-guard (runtime exports == llms.txt == Stream.d.ts, VERSION == package.json version, the declared peer is present in both docs). Each family carries an inline failing control proving the guard can fail.
• 10-pipe-enriched.test.mjs (1.3.0) -- the enriched pipeToSignal: mode/maxBuffer validation, buffer-window snapshots and drop counting, onValue ordering and throw policy, onAbort routing with and without the hook, and the stop-fn droppedCount/overflowCount observability
• 11-writer.test.mjs (1.4.0) -- createSignalWriter: the validation matrix (invalid target / unknown mode / latest+maxBuffer / buffer maxBuffer, including null/-0), terminal idempotence and frozen droppedCount (both end()-first and error()-first orderings), latest-mode zero-drop, the buffer window sequence vs expected arrays, fresh-reference-per-push, the maxBuffer === 1 boundary, undefined/null values, same-reference-twice, re-entrant push()/end() from inside the target's own set(), a throwing target.set(), and the non-callable four-name handle pin

Run via npm test.

Tier 2 -- memory (allocation-free verification)

test/06-gc.test.mjs -- runs under --expose-gc. Asserts heap budget ceilings at 2 MB for the 1.0.0 paths (2K latest, 1K buffer, 1K abort) and the 1.1 additions (5K mode:"latest" resolve, 3K timeout, 1K buffer fill+drain, 2K mode:"latest" churn, 2K filter-throw).

The GC tier now runs under the default npm test (the script carries --expose-gc), so these budgets gate on every run; npm run test:gc is retained as an alias.

Tier 3 -- performance (measured throughput)

bench/bench.mjs -- six scenarios; throughput and B/op retained. Run via npm run bench.

Tier 4 -- torture (retention + alloc gate + controls)

npm run torture (node --expose-gc test/torture.mjs) runs tiers s0-s8 sequentially -- metamorphic laws, degenerate values, protocol conformance, API abuse (with the createSignalWriter validation + terminal no-op rows), seeded fuzz-vs-oracle, the s5 per-scenario allocation gate (maxMajor 0, including the 1.4.0 writerLatest (budget 1 B/op) and writerBuffer64 (budget 32 B/op) scenarios alongside the unchanged pipeSteady default-path budget), a lite-leak retention soak (with the writer create/push/end census), the StreamQuery conformance corpus plus the 1.4.0 writer-vs-pipeToSignal parity contract (the 5 x 6 = 30 m x |V| grid and the 12-case frame-shape corpus, both pinned verbatim from the lite-query handoff), and the controls tier. It prints ok and exits 0 on success. STREAM_TORTURE_BREAK=<s0..s8|s7e|s7w|1> arms one tier's (or every tier's) injected breakage so the control lane exits non-zero; npm run torture:control runs the all-tiers form (11 controls: s0-s8 plus the s7e enriched-surface control and the s7w writer-parity control, each individually non-zero). npm run test:floor runs npm test against the peer floor @zakkster/[email protected] and then npm test + torture against the resolved latest, printing both verdicts.

No package-lock.json is committed: devDependencies float within their stated ranges, and correctness against the peer floor is proven by running npm run test:floor against @zakkster/[email protected] rather than by pinning a lockfile.

What this is not

• Not a full reactive stream library. No map, filter, merge, or any other combinator surface. Compose iterators yourself with async generators -- for await (const x of source) { if (pred(x)) yield x; } -- and pass the composed iterator to fromAsyncIterable.
• Not a backpressure protocol. Async iterators in JS don't have a way to tell the producer "wait, I'm full." The buffer mode's maxBuffer is a ceiling on in-flight values; overflow drops oldest. If you need bidirectional backpressure, your source needs to expose that surface.
• Not a replacement for lite-clock. For frame-rate driven loops, use lite-clock. Use lite-stream for one-way async sources where the source's pace is external.
• Not for one-shot promises. Use lite-await's fromPromise for that. lite-stream is the multi-shot dual; the cleanup model is fundamentally different.

Ecosystem

lite-stream composes with the rest of the @zakkster/lite-* family:

• @zakkster/lite-signal -- the reactive signal core. Required peer.
• @zakkster/lite-await -- single-shot async primitives (whenSignal, withTimeout, withAbort, fromPromise). Pair withTimeout with each next() call to add per-pull timeouts to your iterator.
• @zakkster/lite-statechart -- finite state machines. Use ctx.signal (1.1+) from a state's entry action to drive a lite-stream pump that auto-aborts on state transitions.
• @zakkster/lite-twitch (upcoming) -- Twitch Extension SDK. Helix pagination, EBS SSE, pubsub topics all consume lite-stream.

Bundle

~5 KB minified (gzipped: ~2 KB)
ESM-only, Node >= 18
Single file: Stream.js
Peer dep: @zakkster/lite-signal ^1.2.0
Zero runtime deps.

License

MIT (c) 2026 Zahary Shinikchiev. See LICENSE.txt.