# Capability Spec Delta: streams (enterprise-streams-redesign) **Package**: `@suaveplan/streams` **Status**: PROPOSED > Every requirement carries a `**Verified by:**` line at authoring time. (See `openspec/AGENTS.md` Section 13.) This delta is authored against an empty folded spec (`openspec/specs/streams/spec.md` does not exist on genesis — the prior `enterprise-streams` change archived without folding). All requirements below land as **ADDED** on archive. ## ADDED Requirements ### Requirement: BaseTransform abstract class The package SHALL export an abstract `BaseTransform` class providing the canonical plugin contract for stream transforms. Subclasses implement `process(input: T, options?: TransformOptions): Promise`. Every subclass inherits a uniform hook surface, event surface, parameter introspection, and a `toTransformStream(): TransformStream` bridge to WHATWG Streams. #### Scenario: Subclass implements process and is invokable - **WHEN** a consumer defines `class JsonTransform extends BaseTransform` with `name = "json"` and a concrete `process(input)` that JSON-stringifies the input - **THEN** `new JsonTransform().process({ a: 1 })` returns `'{"a":1}'` #### Scenario: Hooks fire in registration order around process - **WHEN** `t.hook("before", payloadA => payloadA)` and `t.hook("before", payloadB => payloadB)` are registered, then `t.process(...)` runs - **THEN** the first hook receives the original payload and its return value is passed to the second hook before the wrapped operation runs #### Scenario: Events fire without affecting return value - **WHEN** a consumer registers `t.on("complete", payload => sideEffect(payload))` and `process` runs to completion - **THEN** the side effect is observed and `process` still resolves with the operation's actual return value #### Scenario: toTransformStream returns a usable WHATWG TransformStream - **WHEN** a consumer calls `t.toTransformStream()` and pipes a `ReadableStream` through it - **THEN** every chunk is processed by `t.process(chunk)` and the result is enqueued to the readable side #### Scenario: dispose releases resources idempotently - **WHEN** `t.dispose()` is called twice in succession - **THEN** internal hooks and event listeners are cleared on the first call; the second call is a no-op and does not throw **Verified by:** `src/base/transform/transform.test.ts::"subclass process returns transformed value"` AND `src/base/transform/transform.test.ts::"hooks fire in registration order with payload chaining"` AND `src/base/transform/transform.test.ts::"events fire without affecting return value"` AND `src/base/transform/transform.test.ts::"toTransformStream pipes through process"` AND `src/base/transform/transform.test.ts::"dispose is idempotent and clears hooks and listeners"` AND `src/base/transform/transform.spec-conformance.test.ts::"hook payload type is inferred from event name literal"` --- ### Requirement: TransformOptions parameter and Zod schema The package SHALL accept `TransformOptions` on every `process()` call: `{ signal?: AbortSignal; clock?: Clock; logger?: LoggerLike }`. The schema is exported as `TransformOptionsSchema` from a sibling `*.schemas.ts`. #### Scenario: Aborted signal rejects mid-process - **WHEN** `t.process(input, { signal })` runs and `signal.abort(reason)` fires before completion - **THEN** the returned promise rejects with `StreamError("STREAM.ABORTED")` whose `cause` is the abort reason #### Scenario: Schema rejects unknown options at the boundary - **WHEN** `TransformOptionsSchema.parse({ unknownKey: 1 })` is invoked - **THEN** Zod throws with a `ZodError` listing `unknownKey` (strict object schema) **Verified by:** `src/base/transform/transform.test.ts::"aborted signal rejects process with STREAM.ABORTED and propagates cause"` AND `src/base/transform/transform.schemas.test.ts::"TransformOptionsSchema rejects unknown keys"` --- ### Requirement: StreamManager chainable pipeline The package SHALL export `StreamManager` with `pipe(...transforms: BaseTransform[]): StreamManager`, `execute(input: T, options?: ExecuteOptions): Promise`, `clear(): StreamManager`, and `getTransforms(): readonly BaseTransform[]`. #### Scenario: Sequential execution feeds output into next transform - **WHEN** `manager.pipe(a, b).execute(input)` runs - **THEN** `a.process(input)` resolves to `x`, then `b.process(x)` resolves to the final value #### Scenario: Empty pipeline returns input unchanged - **WHEN** `new StreamManager().execute(input)` runs without any pipe - **THEN** the result is exactly `input` (referential equality) #### Scenario: Error in middle transform aborts downstream - **WHEN** transform `b` throws and pipeline is `pipe(a, b, c)` - **THEN** `c.process` is never invoked; the error from `b` propagates to the caller wrapped in `StreamError("STREAM.TRANSFORM_FAILED")` #### Scenario: clear() empties the pipeline - **WHEN** `manager.pipe(a).clear().getTransforms()` is called - **THEN** the returned array has length 0 **Verified by:** `src/manager/manager/manager.test.ts::"pipe chains transforms in order"` AND `src/manager/manager/manager.test.ts::"empty pipeline returns input unchanged"` AND `src/manager/manager/manager.test.ts::"error in middle transform aborts downstream and wraps in STREAM.TRANSFORM_FAILED"` AND `src/manager/manager/manager.test.ts::"clear() empties pipeline"` --- ### Requirement: TransformFactory registry The package SHALL export `TransformFactory` with static `create(type, ...args)` and instance `create(type, ...args)`, where `type` is a literal string keyed against a registry of bundled transforms. Unknown types raise `StreamError("STREAM.UNKNOWN_TRANSFORM_TYPE")`. The registry is open: `TransformFactory.register(type, factory)` adds new types at runtime. #### Scenario: Built-in type resolves to correct subclass instance - **WHEN** `TransformFactory.create("crypto", { engine, key, operation: "encrypt" })` is called - **THEN** the returned instance is `instanceof CryptoTransform` with the supplied options #### Scenario: Unknown type throws structured error - **WHEN** `TransformFactory.create("nonexistent")` is called - **THEN** the call throws `StreamError` with code `"STREAM.UNKNOWN_TRANSFORM_TYPE"` and a message naming `"nonexistent"` #### Scenario: Custom registration is honoured - **WHEN** a consumer calls `TransformFactory.register("upper", () => new UpperCaseTransform())` then `create("upper")` - **THEN** the returned instance is the registered subclass **Verified by:** `src/factory/factory/factory.test.ts::"built-in type resolves to subclass instance"` AND `src/factory/factory/factory.test.ts::"unknown type throws STREAM.UNKNOWN_TRANSFORM_TYPE"` AND `src/factory/factory/factory.test.ts::"custom registration is honoured"` AND `src/factory/factory/factory.spec-conformance.test.ts::"create() narrows return type by literal type argument"` --- ### Requirement: BaseDuplexStream primitive The package SHALL export `BaseDuplexStream` with constructor `(source: StreamSource, sink: StreamSink, options?: DuplexStreamOptions)` exposing `readable: ReadableStream` and `writable: WritableStream`. Supports byte-mode and object-mode via `options.objectMode`. Honours `highWaterMark`. #### Scenario: Bidirectional flow - **WHEN** a consumer writes a chunk to `duplex.writable` and reads from `duplex.readable` whose `pull` is wired to enqueue an echo - **THEN** the echoed chunk is read back #### Scenario: cancel propagates to source - **WHEN** `duplex.readable.cancel(reason)` is called - **THEN** the source's `cancel(reason)` is invoked exactly once with the same reason #### Scenario: Object-mode bypasses byteLength accounting - **WHEN** `objectMode: true` and 1000 non-byte objects are written - **THEN** the writable accepts them subject to `highWaterMark` count, not byte size **Verified by:** `src/primitives/duplex/duplex.test.ts::"bidirectional flow round-trips a chunk"` AND `src/primitives/duplex/duplex.test.ts::"cancel propagates reason to source.cancel"` AND `src/primitives/duplex/duplex.test.ts::"object-mode honours count-based highWaterMark"` --- ### Requirement: NetworkStreamImpl with metadata The package SHALL export `NetworkStreamImpl` extending the duplex pattern with `id: string`, optional `protocol?: string`, `metadata: Map`, `closed: boolean`, `close(): Promise`, and `reset(): Promise`. `close()` cleanly closes both sides; `reset()` aborts both sides with `StreamError("STREAM.NETWORK_RESET")`. #### Scenario: Metadata is accessible after construction - **WHEN** `createNetworkStream("peer-1", "webrtc", new Map([["userId", "u-42"]]))` is called - **THEN** `stream.id === "peer-1"`, `stream.protocol === "webrtc"`, and `stream.metadata.get("userId") === "u-42"` #### Scenario: close() flips closed flag and prevents further writes - **WHEN** `await stream.close()` runs and a subsequent writer attempts to write - **THEN** `stream.closed === true` and the write rejects #### Scenario: reset() aborts with STREAM.NETWORK_RESET - **WHEN** `await stream.reset()` runs while a reader is pending - **THEN** the reader rejects with `StreamError("STREAM.NETWORK_RESET")` **Verified by:** `src/primitives/network/network.test.ts::"metadata id and protocol accessible after construction"` AND `src/primitives/network/network.test.ts::"close() flips closed flag and rejects subsequent writes"` AND `src/primitives/network/network.test.ts::"reset() aborts pending reader with STREAM.NETWORK_RESET"` --- ### Requirement: CryptoTransform with pluggable streaming AEAD The package SHALL export `CryptoTransform` taking `{ engine: StreamingAeadEngine, key: Uint8Array, operation: "encrypt" | "decrypt" }` and exposing the `BaseTransform` plugin surface plus `crypto:before-chunk` and `crypto:after-chunk` hooks and `crypto:start` / `crypto:chunk` / `crypto:complete` / `crypto:error` events. The transform binds an 8-byte counter and 1-byte final-flag into the AEAD AAD, so reorder, truncation, duplication, and tamper all surface as decryption failure. #### Scenario: Encrypt-decrypt round-trip yields identity - **WHEN** the same key is used to construct an `encrypt` and a `decrypt` transform, and a sequence of plaintext frames pipes through both via `toTransformStream()` - **THEN** the output equals the input frame-by-frame #### Scenario: Truncation is detected - **WHEN** an encrypted stream's last frame is dropped before decryption - **THEN** the `decrypt` transform rejects with `StreamError("STREAM.CRYPTO_DECRYPT_FAILED")` referencing AAD verification failure #### Scenario: Reorder is detected - **WHEN** two encrypted middle frames are swapped before decryption - **THEN** the `decrypt` transform rejects with `StreamError("STREAM.CRYPTO_DECRYPT_FAILED")` #### Scenario: Hooks observe and may modify chunks - **WHEN** `t.hook("crypto:before-chunk", p => ({ ...p, chunk: redact(p.chunk) }))` is registered and a frame passes through - **THEN** the redacted chunk is what gets sealed #### Scenario: Events fire in canonical order - **WHEN** an `encrypt` transform processes 3 frames to completion - **THEN** events fire as `crypto:start`, `crypto:chunk` ×3, `crypto:complete` in that order #### Scenario: dispose zeroizes key buffer - **WHEN** `t.dispose()` is called on an encrypt transform - **THEN** the internal copy of the key buffer is overwritten with zeros (via `@suaveplan/crypto-utils.zeroize`) **Verified by:** `src/transforms/crypto/crypto.test.ts::"encrypt-decrypt round-trip preserves frame sequence"` AND `src/transforms/crypto/crypto.test.ts::"truncation produces STREAM.CRYPTO_DECRYPT_FAILED"` AND `src/transforms/crypto/crypto.test.ts::"reorder produces STREAM.CRYPTO_DECRYPT_FAILED"` AND `src/transforms/crypto/crypto.test.ts::"crypto:before-chunk hook can replace plaintext before sealing"` AND `src/transforms/crypto/crypto.test.ts::"events fire in canonical order start/chunk*N/complete"` AND `src/transforms/crypto/crypto.spec-conformance.test.ts::"dispose() zeroizes key buffer"` --- ### Requirement: SplitJoinTransform object-mode chunking The package SHALL export `SplitJoinTransform` with operations `"split"` and `"join"`. Split takes `{ chunkSize, overlap }` and accepts `string`, `ArrayBuffer`, or `T[]`. Join accepts `{ chunks, metadata? }` shape and reconstructs the original. Type guards `isSplittable(value)` and `isJoinable(value)` are exported. #### Scenario: String split with overlap reproduces every character - **WHEN** a 100-char string is split with `chunkSize: 30, overlap: 5` and joined back with `removeOverlap: true` - **THEN** the joined string equals the original #### Scenario: ArrayBuffer split preserves byte-for-byte content - **WHEN** a 1024-byte `ArrayBuffer` is split with `chunkSize: 256, overlap: 0` and joined back - **THEN** the joined buffer equals the original byte-for-byte #### Scenario: Type guard rejects non-splittable input - **WHEN** `isSplittable({ foo: 1 })` is called - **THEN** the result is `false` **Verified by:** `src/transforms/split-join/split-join.test.ts::"string split with overlap round-trips via removeOverlap"` AND `src/transforms/split-join/split-join.test.ts::"ArrayBuffer split round-trips byte-for-byte"` AND `src/transforms/split-join/split-join.test.ts::"isSplittable rejects plain objects"` --- ### Requirement: Functional transform factories preserved The package SHALL preserve the function-form factories from the prior `enterprise-streams` archive: `mapStream`, `filterStream`, `flatMapStream`, `batchStream`, `windowStream`, `throttleStream`, `limitStream`, `dedupStream`, `tapStream`, `encodeTextStream`, `decodeTextStream`, `compressStream`, `decompressStream`, `fromArray`, `fromBytes`, `fromAsyncGenerator`, `toArray`, `toBytes`, `forEach`, `pipeline`, `fanOut`, `merge`, `instrumentTransform`, `fromNodeReadable`, `toNodeReadable`, `fromNodeWritable`. The corresponding `BaseTransform` subclasses delegate to these factories via `toTransformStream()` — no parallel implementation. #### Scenario: MapTransform.toTransformStream() and mapStream(fn) are observably equivalent - **WHEN** the same chunk is piped through `mapStream(fn)` directly and through `new MapTransform({ fn }).toTransformStream()` - **THEN** the output chunks are equal #### Scenario: Functional factory is tree-shakeable - **WHEN** a consumer imports only `mapStream` from `@suaveplan/streams/transforms/map/map` - **THEN** the resulting bundle does not include `BaseTransform` or any unrelated transform **Verified by:** `src/transforms/map/map.test.ts::"MapTransform.toTransformStream and mapStream(fn) produce equivalent output"` AND `tests/tree-shake.test.ts::"importing only mapStream excludes BaseTransform"` --- ### Requirement: ChunkBytes byte-level re-framing The package SHALL export `chunkBytes(frameSize: number, options?: { tail?: 'emit' | 'pad' | 'drop' }): TransformStream` and `ChunkBytesTransform` class. Re-emits any input byte stream as a stream of frames of exactly `frameSize` bytes; the final partial frame is handled per the `tail` policy (default: `'emit'`). #### Scenario: All-equal-size frames concatenate to original input - **WHEN** an input stream of total length `n × frameSize` bytes (chunked arbitrarily) is piped through `chunkBytes(frameSize)` - **THEN** every output frame has length `frameSize` and concatenated output equals input byte-for-byte #### Scenario: Tail policy 'emit' yields a final partial frame - **WHEN** an input of `frameSize × 2 + 7` bytes is piped through `chunkBytes(frameSize, { tail: 'emit' })` - **THEN** the third emitted frame has length 7 and concatenated output equals input #### Scenario: Tail policy 'pad' zero-pads the final frame - **WHEN** the same input is piped through `chunkBytes(frameSize, { tail: 'pad' })` - **THEN** the third frame has length `frameSize` and bytes `frameSize - 7..frameSize - 1` are 0 #### Scenario: Tail policy 'drop' discards incomplete final frame - **WHEN** the same input is piped through `chunkBytes(frameSize, { tail: 'drop' })` - **THEN** only two frames are emitted **Verified by:** `src/transforms/chunk-bytes/chunk-bytes.test.ts::"all frames are exactly frameSize bytes"` AND `src/transforms/chunk-bytes/chunk-bytes.test.ts::"tail emit yields final partial frame and round-trips"` AND `src/transforms/chunk-bytes/chunk-bytes.test.ts::"tail pad zero-pads final frame to frameSize"` AND `src/transforms/chunk-bytes/chunk-bytes.test.ts::"tail drop discards incomplete final frame"` AND `tests/property.test.ts::"chunkBytes byte-conservation under arbitrary input chunking"` --- ### Requirement: CoalesceBytes batching with latency cap The package SHALL export `coalesceBytes(minBytes: number, options?: { maxLatencyMs?: number; clock?: Clock }): TransformStream`. Accumulates incoming chunks until the accumulator reaches `minBytes` or `maxLatencyMs` has elapsed since the first buffered chunk, then emits a single coalesced frame. #### Scenario: Below-threshold inputs are buffered and emitted on close - **WHEN** three 100-byte chunks pipe in (total 300 bytes) with `coalesceBytes(1024)` and the input then closes - **THEN** a single 300-byte frame is emitted before the output closes #### Scenario: Threshold reached emits immediately - **WHEN** chunks accumulate to ≥ `minBytes` mid-stream - **THEN** a single coalesced frame is emitted at that point and the accumulator resets #### Scenario: maxLatencyMs forces an early flush - **WHEN** a single 50-byte chunk is buffered with `coalesceBytes(1024, { maxLatencyMs: 100 })` and 100 ms elapses (FakeClock advance) - **THEN** a 50-byte frame is emitted **Verified by:** `src/transforms/coalesce-bytes/coalesce-bytes.test.ts::"below-threshold buffers and flushes on close"` AND `src/transforms/coalesce-bytes/coalesce-bytes.test.ts::"threshold reached emits coalesced frame and resets"` AND `src/transforms/coalesce-bytes/coalesce-bytes.test.ts::"maxLatencyMs forces early flush via FakeClock"` --- ### Requirement: SlidingBytes overlapping window The package SHALL export `slidingBytes(size: number, stride: number): TransformStream`. Emits overlapping byte windows of length `size` advancing by `stride` bytes per emission. #### Scenario: Adjacent windows overlap by size − stride bytes - **WHEN** a 16-byte input is piped through `slidingBytes(8, 4)` - **THEN** windows are emitted at offsets 0, 4, 8, each of length 8; consecutive windows share their last `size − stride` bytes #### Scenario: Stride larger than size skips bytes - **WHEN** `slidingBytes(4, 8)` runs over a 24-byte input - **THEN** non-overlapping windows are emitted at offsets 0, 8, 16; the 4 bytes between each window are discarded **Verified by:** `src/transforms/sliding-bytes/sliding-bytes.test.ts::"adjacent windows overlap by size minus stride"` AND `src/transforms/sliding-bytes/sliding-bytes.test.ts::"stride larger than size skips bytes between windows"` --- ### Requirement: LengthPrefix encode and decode The package SHALL export `lengthPrefixEncode(prefixBytes: 1 | 2 | 4 | 8, options?: { byteOrder?: 'le' | 'be' })` and a matching `lengthPrefixDecode(prefixBytes, options?: { byteOrder?: 'le' | 'be'; maxFrameBytes?: number })`. Encode prepends a length prefix to each chunk. Decode reads a length prefix, accumulates that many bytes, and emits the framed payload. #### Scenario: Encode-decode round-trip preserves frame boundaries - **WHEN** three frames of varying sizes are piped through `lengthPrefixEncode(4) → lengthPrefixDecode(4)` - **THEN** the three output frames equal the three input frames byte-for-byte #### Scenario: maxFrameBytes overflow throws STREAM.FRAME_TOO_LARGE - **WHEN** decode encounters a length prefix exceeding `maxFrameBytes` - **THEN** the transform errors the readable side with `StreamError("STREAM.FRAME_TOO_LARGE")` containing the prefix value #### Scenario: Truncated frame at close throws STREAM.TRUNCATED_FRAME - **WHEN** the input closes mid-frame with fewer bytes than the declared length prefix - **THEN** the transform errors with `StreamError("STREAM.TRUNCATED_FRAME")` #### Scenario: Big-endian and little-endian round-trip - **WHEN** encode/decode is run with `byteOrder: 'be'` and a frame whose length is `0x01_00` - **THEN** the prefix bytes on the wire are `[0x00, 0x00, 0x01, 0x00]` (4-byte BE) and decode recovers the frame **Verified by:** `src/transforms/length-prefix-encode/length-prefix-encode.test.ts::"encode-decode round-trip preserves three frames"` AND `src/transforms/length-prefix-decode/length-prefix-decode.test.ts::"maxFrameBytes overflow throws STREAM.FRAME_TOO_LARGE"` AND `src/transforms/length-prefix-decode/length-prefix-decode.test.ts::"truncated frame at close throws STREAM.TRUNCATED_FRAME"` AND `src/transforms/length-prefix-encode/length-prefix-encode.test.ts::"big-endian prefix encoding produces expected bytes"` --- ### Requirement: Partition single-pass split The package SHALL export `partition(source: ReadableStream, predicate: (chunk: T) => boolean): { truthy: ReadableStream; falsy: ReadableStream }`. Each input chunk is routed to exactly one of the two output streams (no duplication, unlike `tee`/`fanOut`). Backpressure on either output applies to the source. #### Scenario: Union of outputs equals input, in order per side - **WHEN** `partition(source, n => n % 2 === 0)` is applied to `[1,2,3,4,5,6]` - **THEN** the truthy stream yields `[2,4,6]` and the falsy stream yields `[1,3,5]` #### Scenario: Slow consumer on one side applies backpressure to source - **WHEN** the truthy consumer is held while the source has 1000 items - **THEN** `source` is read at most up to the implementation's high-water-mark before pulls block **Verified by:** `src/pipeline/partition/partition.test.ts::"union of outputs equals input with per-side ordering preserved"` AND `src/pipeline/partition/partition.test.ts::"slow consumer applies backpressure to source"` AND `tests/property.test.ts::"partition order preservation per side under arbitrary inputs"` --- ### Requirement: Route N-way single-pass split The package SHALL export `route(source, keyFn, keys, options?: { onUnknownKey?: 'drop' | 'error' }): Record>` and a non-string-key variant `routeBy(source, keyFn, keys, options?): Map>`. #### Scenario: Each chunk goes to exactly one keyed branch - **WHEN** `route(source, c => c.kind, ["a", "b"])` is applied - **THEN** every input chunk arrives in exactly one branch matching its `kind` #### Scenario: Unknown key with onUnknownKey 'drop' silently discards - **WHEN** a chunk maps to a key not in `keys` and `onUnknownKey: 'drop'` - **THEN** the chunk is consumed from source but emitted nowhere #### Scenario: Unknown key with onUnknownKey 'error' fails the source - **WHEN** a chunk maps to a key not in `keys` and `onUnknownKey: 'error'` - **THEN** every output stream errors with `StreamError("STREAM.ROUTE_UNKNOWN_KEY")` containing the offending key **Verified by:** `src/pipeline/route/route.test.ts::"each chunk routed to exactly one keyed branch"` AND `src/pipeline/route/route.test.ts::"unknown key with drop policy silently discards"` AND `src/pipeline/route/route.test.ts::"unknown key with error policy errors all branches with STREAM.ROUTE_UNKNOWN_KEY"` AND `src/pipeline/route/route.spec-conformance.test.ts::"route() return type is Record keyed by literal keys argument"` --- ### Requirement: Zip sample-aligned merge The package SHALL export `zip(streams: { [K in keyof T]: ReadableStream }, options?: { onShortInput?: 'close' | 'error' | 'pad'; padValue?: unknown }): ReadableStream`. Emits one tuple per round once every input has produced a chunk. #### Scenario: Equal-length streams zip to tuples - **WHEN** `zip([fromArray([1,2,3]), fromArray(["a","b","c"])])` is collected - **THEN** the result is `[[1,"a"], [2,"b"], [3,"c"]]` #### Scenario: Short input with onShortInput 'close' ends output at first end - **WHEN** one input stream ends after 2 chunks, the other after 5, with default `onShortInput: 'close'` - **THEN** the zipped output emits 2 tuples and closes #### Scenario: Short input with onShortInput 'error' rejects - **WHEN** any input ends before others with `onShortInput: 'error'` - **THEN** the zipped output errors with `StreamError("STREAM.ZIP_LENGTH_MISMATCH")` **Verified by:** `src/pipeline/zip/zip.test.ts::"equal-length streams zip to tuples"` AND `src/pipeline/zip/zip.test.ts::"short input with close policy ends at first end"` AND `src/pipeline/zip/zip.test.ts::"short input with error policy throws STREAM.ZIP_LENGTH_MISMATCH"` AND `tests/property.test.ts::"zip then unzip round-trips for equal-length inputs"` --- ### Requirement: Concat sequential drain The package SHALL export `concat(streams: ReadableStream[]): ReadableStream`. Drains each stream to completion in order before pulling from the next. #### Scenario: Order is preserved across boundaries - **WHEN** `concat([fromArray([1,2,3]), fromArray([4,5,6])])` is collected - **THEN** the result is `[1,2,3,4,5,6]` #### Scenario: Error in one stream propagates and halts - **WHEN** the second of three streams errors mid-drain - **THEN** the third stream is never read; the error propagates wrapped in `StreamError("STREAM.PIPE_FAILED")` **Verified by:** `src/pipeline/concat/concat.test.ts::"order preserved across stream boundaries"` AND `src/pipeline/concat/concat.test.ts::"error in middle stream halts and propagates STREAM.PIPE_FAILED"` --- ### Requirement: MergeOrdered k-way merge of sorted streams The package SHALL export `mergeOrdered(streams: ReadableStream[], compare: (a: T, b: T) => number): ReadableStream`. Performs a k-way merge of pre-sorted input streams. #### Scenario: Three sorted streams merge to a globally sorted output - **WHEN** three streams `[1,4,7]`, `[2,5,8]`, `[3,6,9]` are merged with ascending compare - **THEN** the output is `[1,2,3,4,5,6,7,8,9]` #### Scenario: Output is sorted iff inputs are sorted - **WHEN** any input is unsorted - **THEN** the output may not be sorted (documented limitation; not validated at runtime) **Verified by:** `src/pipeline/merge-ordered/merge-ordered.test.ts::"three sorted streams merge to globally sorted output"` AND `tests/property.test.ts::"mergeOrdered output is sorted when all inputs are sorted"` --- ### Requirement: SafeBranch error isolation The package SHALL export `safeBranch(factory: (input: ReadableStream) => ReadableStream, options?: { bus?: EventBusLike; onError?: (err: unknown) => void }): (input: ReadableStream) => ReadableStream`. Wraps a sub-pipeline so that errors inside it do not propagate up; instead they emit `streams:branch-error` on the bus and/or invoke `onError`. #### Scenario: Branch error becomes a bus event without propagation - **WHEN** a branch factory throws on the third chunk and `bus` is provided - **THEN** the bus receives `streams:branch-error` with the error; the outer pipeline does not error #### Scenario: Output stream closes cleanly after branch error - **WHEN** the branch errors and the wrapped factory's output stream is being read - **THEN** the read completes (close) without throwing and no further chunks are emitted **Verified by:** `src/pipeline/safe-branch/safe-branch.test.ts::"branch error emits bus event without propagating to outer pipeline"` AND `src/pipeline/safe-branch/safe-branch.test.ts::"output closes cleanly after branch error"` --- ### Requirement: ToAsyncIterable sink The package SHALL export `toAsyncIterable(stream: ReadableStream, signal?: AbortSignal): AsyncIterable`. Yields chunks until the stream closes or `signal` aborts. #### Scenario: All chunks are yielded in order - **WHEN** `toAsyncIterable(fromArray([1,2,3]))` is iterated with `for await` - **THEN** the loop sees `1, 2, 3` in order #### Scenario: AbortSignal stops iteration - **WHEN** `signal.abort()` fires mid-iteration - **THEN** the iterator's next `next()` rejects with `StreamError("STREAM.ABORTED")` **Verified by:** `src/sinks/to-async-iterable/to-async-iterable.test.ts::"yields chunks in order"` AND `src/sinks/to-async-iterable/to-async-iterable.test.ts::"abort signal stops iteration with STREAM.ABORTED"` --- ### Requirement: Buffer utilities The package SHALL export `getDataFromFile(filePath: string): Promise` (Node-only with runtime env detection), `bufferToUint8Array(buffer: Buffer | ArrayBuffer): Uint8Array`, and `createUint8ArraySlice(array: Uint8Array, start: number, end: number): Uint8Array`. #### Scenario: getDataFromFile reads a Node file as Uint8Array - **WHEN** in Node environment, `await getDataFromFile(path)` is called on an existing 4-byte file with bytes `[1,2,3,4]` - **THEN** the result is a `Uint8Array` of length 4 with bytes `[1,2,3,4]` #### Scenario: getDataFromFile throws in browser environment - **WHEN** `globalThis.process` is undefined and `getDataFromFile(path)` is called - **THEN** the call rejects with `StreamError("STREAM.FILE_BROWSER_NOT_SUPPORTED")` #### Scenario: getDataFromFile throws on missing file - **WHEN** `await getDataFromFile("/no/such/file")` is called in Node - **THEN** the call rejects with `StreamError("STREAM.FILE_READ_FAILED")` whose `cause` is the underlying ENOENT error #### Scenario: bufferToUint8Array preserves bytes from a Node Buffer - **WHEN** `bufferToUint8Array(Buffer.from([0x10, 0x20]))` is called in Node - **THEN** the returned `Uint8Array` has bytes `[0x10, 0x20]` #### Scenario: createUint8ArraySlice produces an independent copy - **WHEN** a slice is created and mutated - **THEN** the original array is unchanged **Verified by:** `src/utils/buffer/get-data-from-file/get-data-from-file.test.ts::"reads a Node file as Uint8Array"` AND `src/utils/buffer/get-data-from-file/get-data-from-file.test.ts::"throws STREAM.FILE_BROWSER_NOT_SUPPORTED when process is undefined"` AND `src/utils/buffer/get-data-from-file/get-data-from-file.test.ts::"throws STREAM.FILE_READ_FAILED with ENOENT cause for missing path"` AND `src/utils/buffer/buffer-to-uint8/buffer-to-uint8.test.ts::"preserves bytes from a Node Buffer"` AND `src/utils/buffer/uint8-slice/uint8-slice.test.ts::"slice mutation does not affect original"` --- ### Requirement: Stream utilities The package SHALL export `streamToUint8Array(stream: ReadableStream, signal?: AbortSignal): Promise`, `uint8ArrayToStream(data: Uint8Array, options?: { chunkSize?: number }): ReadableStream`, `createDataStream(data: T, transformer: (data: T) => Uint8Array | Promise, options?: { chunkSize?: number }): Promise>`, `processDataStream(stream: ReadableStream, processor: (data: Uint8Array) => T | Promise): Promise`, and `validateStreamIntegrity(stream: ReadableStream, validate: (chunk: Uint8Array) => boolean): Promise`. #### Scenario: Round-trip identity - **WHEN** `streamToUint8Array(uint8ArrayToStream(input))` is awaited - **THEN** the result equals `input` byte-for-byte #### Scenario: validateStreamIntegrity returns false on first invalid chunk - **WHEN** any chunk fails the predicate - **THEN** the call resolves to `false` without consuming further chunks **Verified by:** `src/utils/stream/round-trip/round-trip.test.ts::"streamToUint8Array of uint8ArrayToStream is identity"` AND `src/utils/stream/validate-integrity/validate-integrity.test.ts::"returns false on first invalid chunk and short-circuits"` --- ### Requirement: Instrumented transform with OTel The package SHALL export `instrumentTransform(transform: TransformStream, name: string, options?: InstrumentOptions): TransformStream`. Wraps the input transform so each chunk produces a span `stream.transform.`, increments counter `stream.chunks_total`, and records histogram `stream.transform_duration_ms`. Attributes include `stream.stage = name` plus user-supplied `options.attributes`. #### Scenario: Each chunk emits a span and counter increment - **WHEN** an in-memory `MeterProvider` is registered and 5 chunks pass through an instrumented map transform named `"upper"` - **THEN** `stream.chunks_total` is incremented 5 times with attribute `stream.stage = "upper"`; 5 spans named `stream.transform.upper` are emitted #### Scenario: Duration histogram records non-negative values - **WHEN** chunks pass through an instrumented transform that sleeps 10 ms via `FakeClock` - **THEN** every recorded `stream.transform_duration_ms` value is ≥ 0 (FakeClock returns 0; real clock returns positive) **Verified by:** `src/instrument/transform/transform.test.ts::"each chunk emits a span and increments stream.chunks_total"` AND `src/instrument/transform/transform.spec-conformance.test.ts::"in-memory MeterProvider observes counter and histogram emissions"` --- ### Requirement: Node interop adapters The package SHALL export `fromNodeReadable(readable: NodeReadable): ReadableStream`, `toNodeReadable(stream: ReadableStream): NodeReadable`, and `fromNodeWritable(writable: NodeWritable): WritableStream` from the `@suaveplan/streams/node/adapters/adapters` subpath. #### Scenario: Node Readable round-trips through WHATWG - **WHEN** a Node `Readable` emitting `[Buffer.from("a"), Buffer.from("b")]` is converted via `fromNodeReadable`, drained, and re-emitted via `toNodeReadable` - **THEN** the final Node Readable yields the bytes of `"ab"` #### Scenario: Node Writable receives all chunks - **WHEN** a WHATWG `WritableStream` returned by `fromNodeWritable` is fed three chunks - **THEN** the underlying Node Writable observes three `write` calls in order **Verified by:** `src/node/adapters/adapters.test.ts::"Node Readable round-trips through fromNodeReadable and toNodeReadable"` AND `src/node/adapters/adapters.test.ts::"fromNodeWritable forwards chunks to underlying Writable"` --- ### Requirement: Structured error class with stable code registry The package SHALL export `StreamError extends SuaveplanError` and a stable code registry registered via `defineCodes`. Codes: - `STREAM.ABORTED`, `STREAM.BACKPRESSURE_EXCEEDED`, `STREAM.CHUNK_INVALID`, `STREAM.PIPE_FAILED`, `STREAM.TRANSFORM_FAILED`, `STREAM.SOURCE_EXHAUSTED`, `STREAM.READ_FAILED`, `STREAM.WRITE_FAILED`, `STREAM.CLOSED` - `STREAM.FRAME_TOO_LARGE`, `STREAM.TRUNCATED_FRAME`, `STREAM.ROUTE_UNKNOWN_KEY`, `STREAM.ZIP_LENGTH_MISMATCH`, `STREAM.BRANCH_FAILED` - `STREAM.CRYPTO_DECRYPT_FAILED`, `STREAM.CRYPTO_INVALID_FRAME` - `STREAM.DUPLEX_CLOSED`, `STREAM.NETWORK_RESET` - `STREAM.FILE_READ_FAILED`, `STREAM.FILE_BROWSER_NOT_SUPPORTED` - `STREAM.UNKNOWN_TRANSFORM_TYPE` #### Scenario: Every error code is thrown by at least one production path - **WHEN** the code registry is enumerated and cross-checked against grep of `throw new StreamError` - **THEN** every code appears in at least one production-path throw site (verified by `tests/error-coverage.test.ts`) #### Scenario: Subclass narrows code union - **WHEN** a consumer narrows `err.code === "STREAM.ABORTED"` with TypeScript - **THEN** the narrowed branch type is the literal `"STREAM.ABORTED"` **Verified by:** `tests/error-coverage.test.ts::"every StreamErrorCode appears in at least one production throw site"` AND `src/errors/error/error.spec-conformance.test.ts::"narrowing err.code preserves literal type"` --- ### Requirement: Subpath exports per feature module The `package.json` `exports` map SHALL include one entry per feature module so consumers can tree-shake. The barrel `"."` is preserved for ergonomic top-level imports. #### Scenario: Every feature directory under src/ has a corresponding subpath export - **WHEN** `tests/dep-graph.test.ts` enumerates `src/**/index.ts` and matches against `package.json` `exports` - **THEN** every feature index has a matching subpath export #### Scenario: Side-effect-free imports - **WHEN** the entry point `src/index.ts` is loaded - **THEN** no side effects (OTel registration, global mutation) occur (verified via `tests/tree-shake.test.ts`) **Verified by:** `tests/dep-graph.test.ts::"every feature index has a matching subpath export in package.json"` AND `tests/tree-shake.test.ts::"entry-point load produces no observable side effects"` --- ### Requirement: 100% coverage threshold The `vitest.config.ts` SHALL set `thresholds: { statements: 100, branches: 100, functions: 100, lines: 100 }` with `provider: 'istanbul'`. The pre-publish gate `bunx turbo test:coverage --filter=@suaveplan/streams` SHALL hard-fail below 100% on any of the four metrics. #### Scenario: Coverage configuration is set - **WHEN** the vitest config is parsed - **THEN** all four thresholds equal 100 and provider is `'istanbul'` **Verified by:** `tests/dep-graph.test.ts::"vitest.config.ts coverage thresholds are 100/100/100/100 with istanbul provider"` --- ### Requirement: AbortSignal propagation end-to-end The package SHALL accept an `AbortSignal` on every async public entry point: `process`, `execute`, `toArray`, `toBytes`, `forEach`, `pipeline.collect`, `pipeline.drain`, `pipeline.forEach`, `streamToUint8Array`, `validateStreamIntegrity`, `toAsyncIterable`. On abort, in-flight operations cancel underlying readers/writers and reject pending promises with `StreamError("STREAM.ABORTED")`. #### Scenario: Aborting collect cancels source - **WHEN** `pipeline(fromArray([1,2,3,4,5]).pipeThrough(throttle({ everyMs: 100 }))).collect(signal)` runs and `signal.abort()` fires after 50 ms (FakeClock) - **THEN** the promise rejects with `STREAM.ABORTED` and the source has its `cancel()` invoked **Verified by:** `tests/integration.test.ts::"abort during pipeline.collect cancels source and rejects with STREAM.ABORTED"` AND `tests/integration.test.ts::"abort during streamToUint8Array cancels reader"`