suaveplan/openspec/changes/enterprise-streams-redesign/specs/streams/spec.md

# 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<T>(input: T, options?: TransformOptions): Promise<T>`. Every subclass inherits a uniform hook surface, event surface, parameter introspection, and a `toTransformStream<TIn, TOut>(): TransformStream<TIn, TOut>` 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<T>(input: T, options?: ExecuteOptions): Promise<T>`, `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<R, W>` with constructor `(source: StreamSource<R>, sink: StreamSink<W>, options?: DuplexStreamOptions)` exposing `readable: ReadableStream<R>` and `writable: WritableStream<W>`. 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<string, unknown>`, `closed: boolean`, `close(): Promise<void>`, and `reset(): Promise<void>`. `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<Uint8Array, Uint8Array>` 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<Uint8Array, Uint8Array>`. 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<Uint8Array, Uint8Array>`. 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<T>(source: ReadableStream<T>, predicate: (chunk: T) => boolean): { truthy: ReadableStream<T>; falsy: ReadableStream<T> }`. 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<T, K extends string>(source, keyFn, keys, options?: { onUnknownKey?: 'drop' | 'error' }): Record<K, ReadableStream<T>>` and a non-string-key variant `routeBy<T, K>(source, keyFn, keys, options?): Map<K, ReadableStream<T>>`.

#### 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<T extends readonly unknown[]>(streams: { [K in keyof T]: ReadableStream<T[K]> }, options?: { onShortInput?: 'close' | 'error' | 'pad'; padValue?: unknown }): ReadableStream<T>`. 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<T>(streams: ReadableStream<T>[]): ReadableStream<T>`. 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<T>(streams: ReadableStream<T>[], compare: (a: T, b: T) => number): ReadableStream<T>`. 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<TIn, TOut>(factory: (input: ReadableStream<TIn>) => ReadableStream<TOut>, options?: { bus?: EventBusLike; onError?: (err: unknown) => void }): (input: ReadableStream<TIn>) => ReadableStream<TOut>`. 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<T>(stream: ReadableStream<T>, signal?: AbortSignal): AsyncIterable<T>`. 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<Uint8Array>` (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<Uint8Array>, signal?: AbortSignal): Promise<Uint8Array>`, `uint8ArrayToStream(data: Uint8Array, options?: { chunkSize?: number }): ReadableStream<Uint8Array>`, `createDataStream<T>(data: T, transformer: (data: T) => Uint8Array | Promise<Uint8Array>, options?: { chunkSize?: number }): Promise<ReadableStream<Uint8Array>>`, `processDataStream<T>(stream: ReadableStream<Uint8Array>, processor: (data: Uint8Array) => T | Promise<T>): Promise<T>`, and `validateStreamIntegrity(stream: ReadableStream<Uint8Array>, validate: (chunk: Uint8Array) => boolean): Promise<boolean>`.

#### 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<TIn, TOut>(transform: TransformStream<TIn, TOut>, name: string, options?: InstrumentOptions): TransformStream<TIn, TOut>`. Wraps the input transform so each chunk produces a span `stream.transform.<name>`, 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<Uint8Array>`, `toNodeReadable(stream: ReadableStream<Uint8Array>): NodeReadable`, and `fromNodeWritable(writable: NodeWritable): WritableStream<Uint8Array>` 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<StreamErrorCode>` 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"`