first commit

docs/foundation/flow/streams/src/pipeline/builder/builder.md

@@ -0,0 +1,90 @@
+# pipeline-builder
+
+<!-- begin:auto-docs -->
+_(no submodules)_
+<!-- end:auto-docs -->
+## Purpose
+
+`PipelineBuilder` is a fluent, type-safe wrapper for composing WHATWG stream pipelines. Starting from a `ReadableStream<T>`, each call to `.pipe()` returns a new `PipelineBuilder<TNext>` typed to the transform's output, making the full pipeline statically typed from source to sink. Terminal methods (`collect`, `drain`, `forEach`, `drainToWritable`) consume the pipeline and return a Promise.
+
+The builder is a convenience layer over `ReadableStream.pipeThrough` — it does not buffer, copy, or re-implement backpressure. Every `pipe()` call delegates directly to the WHATWG `pipeThrough` method.
+
+## Features
+
+- Fluent `.pipe(transform)` and `.pipe(fn)` chaining with full TypeScript inference
+- Terminal: `collect()`, `drain()`, `forEach()`, `drainToWritable()`
+- `toReadable()` escape hatch for raw stream access
+- Optional global `AbortSignal` in `PipelineOptions` applied to all terminals
+- Per-terminal signal override
+
+## Usage: Basic Example
+
+```ts
+import { pipeline } from "@suaveplan/streams/pipeline/pipeline-builder/pipeline-builder";
+import { fromArray } from "@suaveplan/streams/readable/from-array/from-array";
+import { mapStream } from "@suaveplan/streams/transforms/map-stream/map-stream";
+import { filterStream } from "@suaveplan/streams/transforms/filter-stream/filter-stream";
+
+const result = await pipeline(fromArray([1, 2, 3, 4, 5]))
+  .pipe(filterStream((n) => n % 2 === 0))
+  .pipe(mapStream((n) => n * 10))
+  .collect();
+
+console.log(result); // [20, 40]
+```
+
+## Usage: Advanced Example
+
+```ts
+import { pipeline } from "@suaveplan/streams/pipeline/pipeline-builder/pipeline-builder";
+import { fromArray } from "@suaveplan/streams/readable/from-array/from-array";
+
+const ac = new AbortController();
+
+const result = await pipeline(fromArray(["hello", "world"]), { signal: ac.signal })
+  .pipe((s) => s.toUpperCase())           // inline function
+  .pipe((s) => `[${s}]`)                  // another inline function
+  .collect();
+
+console.log(result); // ["[HELLO]", "[WORLD]"]
+
+// Drain to writable
+const received: string[] = [];
+const writable = new WritableStream<string>({ write(c) { received.push(c); } });
+await pipeline(fromArray(["a", "b"])).drainToWritable(writable);
+console.log(received); // ["a", "b"]
+```
+
+## API Reference
+
+### `pipeline<T>(source: ReadableStream<T>, options?: PipelineOptions): PipelineBuilder<T>`
+
+Creates a new builder from `source`.
+
+### `PipelineBuilder<T>.pipe<TNext>(transform): PipelineBuilder<TNext>`
+
+Applies a `TransformStream<T, TNext>` or a mapping function `(chunk: T) => TNext | Promise<TNext>`.
+
+### `PipelineBuilder<T>.collect(signal?): Promise<T[]>`
+
+Materialises all output chunks into an array.
+
+### `PipelineBuilder<T>.drain(signal?): Promise<void>`
+
+Discards all output chunks.
+
+### `PipelineBuilder<T>.forEach(fn, signal?): Promise<void>`
+
+Calls `fn` for each output chunk.
+
+### `PipelineBuilder<T>.drainToWritable(writable, signal?): Promise<void>`
+
+Pipes all output to a `WritableStream<T>`.
+
+### `PipelineBuilder<T>.toReadable(): ReadableStream<T>`
+
+Returns the underlying `ReadableStream` for raw access.
+
+## Implementation Notes
+
+The inline function overload of `pipe` wraps the function in a `new TransformStream` with a single `transform` callback. This is identical in behaviour to `mapStream` but allows callers to write terse one-liners. Type inference works because the overload signatures have distinct type parameters and TypeScript's overload resolution picks the correct one based on the argument type.

docs/foundation/flow/streams/src/pipeline/concat/concat.md

@@ -0,0 +1,113 @@
+# concat
+
+<!-- begin:auto-docs -->
+_(no submodules)_
+<!-- end:auto-docs -->
+## Purpose
+
+`concat` joins multiple `ReadableStream<T>` inputs end-to-end into a single
+`ReadableStream<T>`, draining each source to completion before pulling from the
+next. This is the sequential counterpart to `merge`: where `merge` races readers
+and emits chunks in arrival order, `concat` preserves the input ordering
+deterministically. Use it when downstream consumers depend on the relative
+order of upstream chunks — for example, replaying a header stream followed by a
+body stream, or stitching paginated results into one contiguous flow.
+
+## Features
+
+- Deterministic, sequential ordering across stream boundaries.
+- Lazy: the next source is only acquired (its reader pulled) once the previous
+  one has reported `done`.
+- Backpressure-aware: pulls from the current source only when the output is
+  being read.
+- Error propagation: any input error halts the concatenation immediately and is
+  surfaced as a `StreamPipeFailedError` (`STREAM.PIPE_FAILED`) with the original
+  error attached as `cause`. Unread later sources are never touched.
+- Cancellation: cancelling the output cancels the currently-reading source.
+  Sources that have not yet been acquired are left untouched (the caller still
+  owns them).
+- Fast paths for empty input (`[]` returns an immediately-closed stream) and
+  single input (returned verbatim, identity).
+
+## Usage: Basic Example
+
+```ts
+import { concat } from "@suaveplan/streams/pipeline/concat/concat";
+import { fromArray } from "@suaveplan/streams/readable/from-array/from-array";
+import { toArray } from "@suaveplan/streams/sinks/to-array/to-array";
+
+const head = fromArray([1, 2, 3]);
+const tail = fromArray([4, 5, 6]);
+
+const all = await toArray(concat([head, tail]));
+// → [1, 2, 3, 4, 5, 6]
+```
+
+## Usage: Advanced Example
+
+Concatenating paginated fetch results, surfacing transport errors:
+
+```ts
+import { concat } from "@suaveplan/streams/pipeline/concat/concat";
+import { StreamPipeFailedError } from "@suaveplan/streams/errors/stream-error";
+
+function pageStream(pageNumber: number): ReadableStream<Uint8Array> {
+    return new ReadableStream({
+        async pull(controller) {
+            const res = await fetch(`/api/items?page=${pageNumber}`);
+            if (!res.ok) throw new Error(`page ${pageNumber} failed`);
+            const buf = new Uint8Array(await res.arrayBuffer());
+            controller.enqueue(buf);
+            controller.close();
+        },
+    });
+}
+
+const stitched = concat([pageStream(1), pageStream(2), pageStream(3)]);
+
+try {
+    const reader = stitched.getReader();
+    while (true) {
+        const { done, value } = await reader.read();
+        if (done) break;
+        process(value);
+    }
+} catch (err) {
+    if (err instanceof StreamPipeFailedError) {
+        // err.cause holds the original transport / page error
+        report(err);
+    }
+    throw err;
+}
+```
+
+## API Reference
+
+### `concat<T>(streams: ReadableStream<T>[]): ReadableStream<T>`
+
+- **Parameters**
+  - `streams` — an ordered array of `ReadableStream<T>` to drain in sequence.
+- **Returns** — a new `ReadableStream<T>` that emits every chunk from
+  `streams[0]`, then every chunk from `streams[1]`, and so on, closing once the
+  final source closes. For `streams.length === 0`, returns an
+  immediately-closed stream. For `streams.length === 1`, returns
+  `streams[0]` unchanged (identity).
+- **Throws (downstream)** — `StreamPipeFailedError` with stable code
+  `STREAM.PIPE_FAILED` whenever any input source errors. The original error is
+  attached as `cause`.
+
+## Implementation Notes
+
+- The implementation owns at most one reader at a time. After a source reports
+  `done`, its reader is `releaseLock`-ed before the next source's reader is
+  acquired. This avoids holding multiple locks open and keeps memory bounded.
+- Sources are acquired lazily inside `pull`. Streams that the consumer never
+  reaches (e.g., because of an early error or cancellation) are never read,
+  which is intentional — concat must not eagerly drain inputs.
+- Cancellation only cancels the currently-acquired source. Un-acquired sources
+  are still owned by the caller and may be reused or cancelled independently.
+- Errors are wrapped, never re-thrown raw. This guarantees that consumers can
+  reliably pattern-match on `StreamPipeFailedError` / `STREAM.PIPE_FAILED` for
+  any concat-mediated failure, regardless of the underlying cause.
+- The single-source identity fast path is observable via reference equality —
+  consumers can detect it with `concat([s]) === s`.

docs/foundation/flow/streams/src/pipeline/fan-out/fan-out.md

@@ -0,0 +1,68 @@
+# fan-out
+
+<!-- begin:auto-docs -->
+_(no submodules)_
+<!-- end:auto-docs -->
+## Purpose
+
+`fanOut` splits a single `ReadableStream<T>` into N independent readable streams that each receive every chunk in the same order. It is built on top of the WHATWG `ReadableStream.tee()` method, applied recursively to produce N branches. All branches are synchronised: the source stream is only pulled when the slowest consumer is ready, so backpressure from any branch applies to all.
+
+Fan-out is the foundation for broadcast patterns: send the same data to a logger, a metrics collector, and a storage writer simultaneously, without running the source pipeline multiple times.
+
+## Features
+
+- N-way broadcast of a single stream
+- Built on native `tee()` — no custom buffering
+- Backpressure from slowest consumer propagates to source
+- Throws `RangeError` for `n < 1`
+- `n = 1` returns the original stream unchanged
+
+## Usage: Basic Example
+
+```ts
+import { fanOut } from "@suaveplan/streams/pipeline/fan-out/fan-out";
+import { fromArray } from "@suaveplan/streams/readable/from-array/from-array";
+import { toArray } from "@suaveplan/streams/sinks/to-array/to-array";
+
+const [a, b] = fanOut(fromArray([1, 2, 3]), 2);
+const [ra, rb] = await Promise.all([toArray(a), toArray(b)]);
+console.log(ra); // [1, 2, 3]
+console.log(rb); // [1, 2, 3]
+```
+
+## Usage: Advanced Example
+
+```ts
+import { fanOut } from "@suaveplan/streams/pipeline/fan-out/fan-out";
+import { fromArray } from "@suaveplan/streams/readable/from-array/from-array";
+import { mapStream } from "@suaveplan/streams/transforms/map-stream/map-stream";
+import { toArray } from "@suaveplan/streams/sinks/to-array/to-array";
+
+const [forStorage, forLogging, forMetrics] = fanOut(
+  fromArray([{ id: 1, value: 42 }, { id: 2, value: 99 }]),
+  3
+);
+
+const [stored, logged, counted] = await Promise.all([
+  toArray(forStorage),
+  toArray(forLogging.pipeThrough(mapStream((e) => `[LOG] id=${e.id}`))),
+  toArray(forMetrics.pipeThrough(mapStream((e) => e.value))),
+]);
+```
+
+## API Reference
+
+### `fanOut<T>(stream: ReadableStream<T>, n: number): ReadableStream<T>[]`
+
+**Parameters**:
+
+- `stream` — the source stream to fan out.
+- `n` — number of output branches; must be `>= 1`.
+
+**Returns**: Array of `n` `ReadableStream<T>` instances.
+
+**Throws** (immediate): `RangeError` if `n < 1`.
+
+## Implementation Notes
+
+`tee()` is called `n - 1` times, each time on the last branch, building a chain: `[a1, (tee b1 into [a2, (tee b2 into [a3, ...])])]`. Each `tee` call creates internal queues that buffer chunks until both branches have consumed them. For large `n`, memory usage grows proportionally to `n × backlog_size`. If consumers process at very different rates, the internal queues can grow unboundedly — combine with `limitStream` or external flow control if this is a concern.

docs/foundation/flow/streams/src/pipeline/merge-ordered/merge-ordered.md

@@ -0,0 +1,119 @@
+# mergeOrdered
+
+<!-- begin:auto-docs -->
+_(no submodules)_
+<!-- end:auto-docs -->
+## Purpose
+
+`mergeOrdered` performs a k-way merge of pre-sorted `ReadableStream`s
+into a single ordered output stream. It is the streaming analogue of
+the merge step in merge-sort: when every input is sorted under the
+same comparator, the output is sorted under that comparator. This is
+the right tool whenever a pipeline needs to consume multiple already-
+ordered sources — sharded query results, time-keyed event logs,
+externally-sorted partitions — without re-sorting the union in memory.
+
+Unlike the unordered `merge`, which races inputs and emits in arrival
+order, `mergeOrdered` emits the global minimum at every step using a
+binary min-heap of `(value, sourceIdx)` entries. Each pull dequeues
+the heap root, enqueues it on the output, and replaces the slot with
+the next chunk from the same source. Throughput is dominated by the
+`O(log k)` heap operations where `k` is the number of inputs.
+
+## Features
+
+- K-way merge over `ReadableStream<T>` arrays of any cardinality
+- Caller-supplied comparator (ascending, descending, or domain-specific)
+- Single-stream and empty-array fast paths
+- Streams of differing lengths handled without truncation
+- `cancel` propagates to every input stream
+- Errors on any input cancel the remaining inputs and surface on the
+  merged stream
+- Backpressure-respecting — only pulls one chunk per source per pull cycle
+
+## Usage: Basic Example
+
+```ts
+import { mergeOrdered } from "@suaveplan/streams/pipeline/merge-ordered/merge-ordered";
+import { fromArray } from "@suaveplan/streams/readable/from-array/from-array";
+import { toArray } from "@suaveplan/streams/sinks/to-array/to-array";
+
+const s1 = fromArray([1, 4, 7]);
+const s2 = fromArray([2, 5, 8]);
+const s3 = fromArray([3, 6, 9]);
+
+const merged = mergeOrdered([s1, s2, s3], (a, b) => a - b);
+const out = await toArray(merged);
+// out === [1, 2, 3, 4, 5, 6, 7, 8, 9]
+```
+
+## Usage: Advanced Example
+
+Merge sharded events keyed by timestamp, in descending order, and
+forward the result into a downstream consumer that may cancel mid-flight:
+
+```ts
+import { mergeOrdered } from "@suaveplan/streams/pipeline/merge-ordered/merge-ordered";
+
+interface Event {
+    ts: number;
+    payload: unknown;
+}
+
+declare const shardA: ReadableStream<Event>;
+declare const shardB: ReadableStream<Event>;
+declare const shardC: ReadableStream<Event>;
+
+const merged = mergeOrdered<Event>(
+    [shardA, shardB, shardC],
+    (a, b) => b.ts - a.ts, // newest-first
+);
+
+const reader = merged.getReader();
+try {
+    const first = await reader.read();
+    if (!first.done) {
+        // process newest event
+    }
+} finally {
+    // Cancellation propagates to every shard.
+    await reader.cancel("consumer-done");
+}
+```
+
+## API Reference
+
+### `mergeOrdered<T>(streams, compare): ReadableStream<T>`
+
+- `streams: ReadableStream<T>[]` — input streams, each pre-sorted
+  under `compare`. May be empty (returns an immediately-closed stream)
+  or singleton (returns a passthrough merge with the same heap path).
+- `compare: (a: T, b: T) => number` — comparator. Negative when `a`
+  precedes `b`, zero when equal, positive otherwise. Identical contract
+  to `Array.prototype.sort`.
+- **Returns** a `ReadableStream<T>` that emits values from the union of
+  all inputs in `compare` order, then closes.
+- **Throws / errors** — if any input reader rejects, the remaining
+  inputs are cancelled and the rejection is forwarded as a stream error.
+
+### `MergeOrderedCompare<T>`
+
+Type alias for the comparator argument; exported for downstream typing
+of comparator-producing factories.
+
+## Implementation Notes
+
+- The heap stores at most `k` entries (one per active source) so peak
+  memory is `O(k)`, independent of the total chunk count.
+- Heap operations (`siftUp` / `siftDown`) are iterative and avoid
+  recursion to keep call stacks shallow on large `k`.
+- The first `pull` primes every reader in parallel via `Promise.all`;
+  subsequent pulls only read from the source whose value was just
+  emitted, which preserves backpressure on the other inputs.
+- `cancel` on the output uses `Promise.allSettled` to ensure every
+  input is cancelled even if one rejects.
+- Output sortedness is conditional on input sortedness — passing
+  unsorted inputs is a caller bug, not a runtime check, because
+  validating sortedness would defeat the streaming guarantees.
+- The single-stream path still goes through the heap for behavioural
+  parity with k≥2 (cancellation, error propagation).

docs/foundation/flow/streams/src/pipeline/merge/merge.md

@@ -0,0 +1,66 @@
+# merge
+
+<!-- begin:auto-docs -->
+_(no submodules)_
+<!-- end:auto-docs -->
+## Purpose
+
+`merge` combines multiple `ReadableStream<T>` instances into a single `ReadableStream<T>` that emits chunks in arrival order. It is a fan-in operation: the merged stream races all input readers simultaneously and yields whichever chunk becomes available first, interleaving outputs from all sources as they produce data. The merged stream closes when every input stream has closed.
+
+This is useful for consuming multiple independent data sources concurrently — for example, merging results from several parallel API calls, aggregating events from multiple sensors, or combining log streams from different servers.
+
+## Features
+
+- Race-based interleaving — faster streams yield chunks sooner
+- Merged stream closes when all inputs close
+- Error from any input stream errors the merged stream
+- Handles empty input (returns immediately-closed stream)
+- Single-stream input returns the stream unchanged
+
+## Usage: Basic Example
+
+```ts
+import { merge } from "@suaveplan/streams/pipeline/merge/merge";
+import { fromArray } from "@suaveplan/streams/readable/from-array/from-array";
+import { toArray } from "@suaveplan/streams/sinks/to-array/to-array";
+
+const [a, b] = [fromArray([1, 3, 5]), fromArray([2, 4, 6])];
+const merged = merge([a, b]);
+const result = await toArray(merged);
+// result contains 1,2,3,4,5,6 in some interleaved order
+console.log(result.length); // 6
+```
+
+## Usage: Advanced Example
+
+```ts
+import { merge } from "@suaveplan/streams/pipeline/merge/merge";
+import { fromAsyncGenerator } from "@suaveplan/streams/readable/from-generator/from-generator";
+import { toArray } from "@suaveplan/streams/sinks/to-array/to-array";
+
+async function* slow() { yield "slow-1"; await new Promise(r => setTimeout(r, 10)); yield "slow-2"; }
+async function* fast() { yield "fast-1"; yield "fast-2"; }
+
+const result = await toArray(merge([
+  fromAsyncGenerator(slow),
+  fromAsyncGenerator(fast),
+]));
+console.log(result); // fast items likely come before slow items
+```
+
+## API Reference
+
+### `merge<T>(streams: ReadableStream<T>[]): ReadableStream<T>`
+
+**Parameters**:
+
+- `streams` — array of readable streams to merge.
+
+**Returns**: `ReadableStream<T>` that emits all chunks from all inputs.
+
+**Returns immediately-closed stream** if `streams` is empty.
+**Returns `streams[0]` unchanged** if `streams.length === 1`.
+
+## Implementation Notes
+
+Each input reader has an in-flight `read()` promise maintained in a `Map`. On each `pull`, `Promise.race` is called over all active in-flight promises to get the next available chunk. When a reader is done, it is removed from the active set. When all readers are done, the controller is closed. Errors from any reader propagate immediately to the merged stream's error channel, and all other readers are cancelled via `Promise.allSettled` in the `cancel` callback.

docs/foundation/flow/streams/src/pipeline/partition/partition.md

@@ -0,0 +1,105 @@
+# `partition`
+
+<!-- begin:auto-docs -->
+_(no submodules)_
+<!-- end:auto-docs -->
+## Purpose
+
+`partition` splits a single `ReadableStream<T>` into two output streams using a
+boolean predicate. Unlike `tee` or `fanOut`, every input chunk is routed to
+_exactly one_ output — the truthy stream or the falsy stream — so the union of
+the two outputs equals the source with per-side ordering preserved. This is
+the right tool when you want to fork a stream by classification (errors vs
+successes, even vs odd, hot vs cold) without paying the buffering cost of a
+full duplication.
+
+## Features
+
+- Single-pass split via one underlying reader on the source
+- Synchronous or asynchronous predicate (`(chunk: T) => boolean | Promise<boolean>`)
+- Backpressure flows from either output back to the source — pulls on each
+  side drive the reader on demand
+- Predicate exceptions error _both_ output streams and cancel the source
+- Per-side `cancel()` cooperates: the source is cancelled only when both
+  sides have been cancelled
+- One-slot per-side buffer absorbs the off-routing case (a pull on truthy
+  yielding a falsy chunk and vice versa)
+- Zero workspace runtime dependencies — uses only WHATWG Streams primitives
+
+## Usage: Basic Example
+
+```ts
+import { partition } from "@suaveplan/streams/pipeline/partition/partition";
+
+declare const numbers: ReadableStream<number>;
+const { truthy: evens, falsy: odds } = partition(
+    numbers,
+    (n) => n % 2 === 0
+);
+```
+
+## Usage: Advanced Example
+
+```ts
+import { partition } from "@suaveplan/streams/pipeline/partition/partition";
+
+interface Event {
+    level: "info" | "warn" | "error";
+    payload: unknown;
+}
+
+declare const events: ReadableStream<Event>;
+
+const { truthy: errors, falsy: rest } = partition(events, async (e) => {
+    // async predicates are awaited in routing order
+    await Promise.resolve();
+    return e.level === "error";
+});
+
+// Each side is an independent ReadableStream — pipe through unrelated
+// pipelines without worrying about duplication or fan-out cost.
+const errorsToSink = errors.pipeTo(errorSink);
+const restToSink = rest.pipeTo(analyticsSink);
+
+await Promise.all([errorsToSink, restToSink]);
+```
+
+## API Reference
+
+### `partition<T>(source, predicate): { truthy, falsy }`
+
+- **`source: ReadableStream<T>`** — the input stream. `partition` acquires a
+  single default reader and drives it from the output pulls.
+- **`predicate: (chunk: T) => boolean | Promise<boolean>`** — routes each
+  chunk: `true` enqueues to `truthy`, `false` enqueues to `falsy`. Predicate
+  exceptions are caught, both outputs are errored with the original cause,
+  and the source reader is cancelled.
+- **Returns `{ truthy: ReadableStream<T>; falsy: ReadableStream<T> }`** — two
+  independent readable streams. Both close when the source is exhausted.
+  Cancelling one side does _not_ cancel the source; cancelling both does.
+
+### Throws
+
+- Synchronous: none — invalid arguments propagate as TypeErrors from the
+  underlying `ReadableStream` machinery on first use.
+- Asynchronous: predicate exceptions surface as stream errors on both
+  outputs; their `.cause` is the original thrown value.
+
+## Implementation Notes
+
+The implementation owns a single `ReadableStreamDefaultReader` on the source
+and exposes two consumer streams whose `pull()` callbacks both drain from
+that reader. Each `pull` loops until it produces a chunk _for its own side_;
+chunks for the other side are stashed in a one-slot buffer that the other
+side claims on its next pull. Because pulls only fire under backpressure
+(when the consumer side has space), the source is read at most as fast as
+the _currently-pulling_ consumer demands — backpressure on either side is
+preserved end-to-end.
+
+Cancellation is cooperative. A `cancel()` on a single output marks that side
+closed; the source reader is only cancelled once both sides are closed. This
+mirrors `tee` semantics and lets the live side continue draining the source.
+
+Predicate exceptions are routed once: the first throw errors both outputs
+and cancels the reader, after which `pullFor` short-circuits via the
+`errored` flag.

docs/foundation/flow/streams/src/pipeline/route/route.md

@@ -0,0 +1,115 @@
+# route / routeBy
+
+<!-- begin:auto-docs -->
+_(no submodules)_
+<!-- end:auto-docs -->
+## Purpose
+
+`route` and `routeBy` split a single `ReadableStream<T>` into N independent
+output branches, but unlike `fanOut` each chunk is delivered to **exactly one**
+branch — the one selected by a caller-supplied `keyFn`. They are the
+single-pass partitioning primitive of the streams pipeline: a chunk is read
+from the source once and dispatched, never duplicated.
+
+This is the idiomatic way to model demultiplexing: a tagged event stream that
+should be processed differently per kind (writes → DB, audits → log,
+notifications → fan-out), without paying the buffering/backpressure-coupling
+cost of `fanOut`. The two ops differ only in container: `route` returns a
+`Record<K, ReadableStream<T>>` for `string` keys, `routeBy` returns a
+`Map<K, ReadableStream<T>>` for arbitrary key types (numbers, symbols,
+objects).
+
+## Features
+
+- Single-pass: each source chunk is dispatched to exactly one branch.
+- Per-key ordering is preserved (FIFO within each branch).
+- Configurable unknown-key policy: `'drop'` (silent, default) or `'error'`
+  (every branch errors with `StreamRouteUnknownKeyError`).
+- Source errors propagate to every branch.
+- `routeBy` accepts non-string keys (number, symbol, object identity).
+- Zero external dependencies; pure WHATWG Streams.
+
+## Usage: Basic Example
+
+```ts
+import { route } from "@suaveplan/streams/pipeline/route/route";
+import { fromArray } from "@suaveplan/streams/readable/from-array/from-array";
+import { toArray } from "@suaveplan/streams/sinks/to-array/to-array";
+
+const events = fromArray([
+    { kind: "a" as const, n: 1 },
+    { kind: "b" as const, n: 2 },
+    { kind: "a" as const, n: 3 },
+]);
+
+const branches = route(events, (e) => e.kind, ["a", "b"] as const);
+const [as, bs] = await Promise.all([
+    toArray(branches.a),
+    toArray(branches.b),
+]);
+// as -> [{ kind: "a", n: 1 }, { kind: "a", n: 3 }]
+// bs -> [{ kind: "b", n: 2 }]
+```
+
+## Usage: Advanced Example
+
+```ts
+import { routeBy } from "@suaveplan/streams/pipeline/route/route";
+import { fromArray } from "@suaveplan/streams/readable/from-array/from-array";
+import { toArray } from "@suaveplan/streams/sinks/to-array/to-array";
+
+// Symbol-keyed routing with strict unknown-key handling.
+const PRIORITY_HIGH = Symbol("high");
+const PRIORITY_LOW = Symbol("low");
+
+const m = routeBy<{ pri: symbol; payload: string }, symbol>(
+    fromArray([
+        { pri: PRIORITY_HIGH, payload: "urgent" },
+        { pri: PRIORITY_LOW, payload: "later" },
+    ]),
+    (e) => e.pri,
+    [PRIORITY_HIGH, PRIORITY_LOW],
+    { onUnknownKey: "error" },
+);
+
+const high = m.get(PRIORITY_HIGH)!;
+const low = m.get(PRIORITY_LOW)!;
+const [hi, lo] = await Promise.all([toArray(high), toArray(low)]);
+```
+
+## API Reference
+
+### `route<T, K extends string>(source, keyFn, keys, options?)`
+
+- `source: ReadableStream<T>` — input stream to dispatch.
+- `keyFn: (chunk: T) => K` — extracts the routing key from each chunk.
+- `keys: readonly K[]` — declared key set; one branch is created per key.
+- `options.onUnknownKey?: 'drop' | 'error'` — default `'drop'`.
+
+**Returns**: `Record<K, ReadableStream<T>>` — one branch per declared key.
+
+**Throws** (per branch, asynchronously): `StreamRouteUnknownKeyError`
+(`STREAM.ROUTE_UNKNOWN_KEY`) when a chunk's key is not in `keys` and
+`onUnknownKey === 'error'`.
+
+### `routeBy<T, K>(source, keyFn, keys, options?)`
+
+Same contract as `route`, but returns `Map<K, ReadableStream<T>>` so the key
+type is unconstrained (numbers, symbols, objects).
+
+## Implementation Notes
+
+The implementation creates one `ReadableStream<T>` per declared key whose
+controller is captured during `start()`. A single reader loop pulls from the
+source on the next microtask after construction and dispatches each chunk to
+the controller selected by `keyFn`. On unknown key:
+
+- `'drop'` — the chunk is consumed and discarded.
+- `'error'` — every branch's controller is errored with
+  `StreamRouteUnknownKeyError`, the source is cancelled, and the dispatch loop
+  terminates.
+
+Branches use the default WHATWG queuing strategy (CountQueuingStrategy with
+`highWaterMark: 1`); if a downstream consumer falls far behind, the branch's
+internal queue grows. Compose with `limit` or `throttle` if an unbounded
+producer must be guarded.

docs/foundation/flow/streams/src/pipeline/safe-branch/safe-branch.md

@@ -0,0 +1,92 @@
+# safeBranch
+
+<!-- begin:auto-docs -->
+_(no submodules)_
+<!-- end:auto-docs -->
+## Purpose
+
+`safeBranch` is a pipeline operator that wraps a sub-pipeline factory in a fault boundary. It runs the supplied `factory(input)` to construct an inner `ReadableStream`, then forwards every chunk it produces to the wrapper's output stream. If the inner pipeline throws — synchronously while the factory is being invoked, or asynchronously while its stream is being read — the wrapper's output stream closes cleanly (no further chunks emitted) and the error is forwarded to `options.onError` as a `StreamBranchFailedError` whose `cause` is the original error.
+
+The outer pipeline never observes the inner error. This is the difference between a normal `pipeThrough` (which propagates errors and tears down the pipeline) and `safeBranch` (which contains them). Use it for opportunistic side branches — analytics, mirroring, telemetry — where a failure must not bring down the primary data path.
+
+## Features
+
+- Isolates errors from inner sub-pipelines so the outer pipeline keeps running
+- Forwards wrapped errors to `options.onError` as `StreamBranchFailedError`
+- Preserves the original error via the `cause` chain
+- Output stream closes cleanly after a branch error — no half-emitted state
+- Silent mode when `onError` is omitted
+- Synchronous factory throws are caught and reported the same way
+
+## Usage: Basic Example
+
+```ts
+import { fromArray } from "@suaveplan/streams/readable/from-array/from-array";
+import { safeBranch } from "@suaveplan/streams/pipeline/safe-branch/safe-branch";
+import { toArray } from "@suaveplan/streams/sinks/to-array/to-array";
+
+const wrapped = safeBranch<number, number>(
+    (input) =>
+        input.pipeThrough(
+            new TransformStream<number, number>({
+                transform(chunk, c) {
+                    c.enqueue(chunk * 2);
+                },
+            })
+        )
+);
+
+const out = wrapped(fromArray([1, 2, 3]));
+const result = await toArray(out); // [2, 4, 6]
+```
+
+## Usage: Advanced Example
+
+```ts
+import { fromArray } from "@suaveplan/streams/readable/from-array/from-array";
+import { safeBranch } from "@suaveplan/streams/pipeline/safe-branch/safe-branch";
+import { StreamBranchFailedError } from "@suaveplan/streams/errors/stream-error";
+import { toArray } from "@suaveplan/streams/sinks/to-array/to-array";
+
+let lastError: StreamBranchFailedError | undefined;
+
+const analyticsBranch = safeBranch<number, number>(
+    (input) =>
+        input.pipeThrough(
+            new TransformStream<number, number>({
+                transform(chunk, c) {
+                    if (chunk < 0) throw new Error("negative not allowed");
+                    c.enqueue(chunk);
+                },
+            })
+        ),
+    {
+        onError: (err) => {
+            lastError = err as StreamBranchFailedError;
+        },
+    }
+);
+
+const collected = await toArray(analyticsBranch(fromArray([1, -2, 3])));
+// collected captures whatever was emitted before the error;
+// the outer pipeline does not throw.
+console.log(lastError?.code); // "STREAM.BRANCH_FAILED"
+console.log(lastError?.cause); // Error: negative not allowed
+```
+
+## API Reference
+
+### `safeBranch<TIn, TOut>(factory, options?)`
+
+- `factory: (input: ReadableStream<TIn>) => ReadableStream<TOut>` — constructs the inner sub-pipeline. May throw synchronously; failures are isolated.
+- `options.onError?: (err: unknown) => void` — invoked with a `StreamBranchFailedError` when the inner pipeline errors. The original error is exposed via `err.cause`.
+- **Returns** `(input: ReadableStream<TIn>) => ReadableStream<TOut>` — a wrapper function suitable for use anywhere a stream-shape transform is accepted.
+- **Throws** Never. All errors are converted to `StreamBranchFailedError` and either delivered to `onError` or silently swallowed.
+
+## Implementation Notes
+
+- The wrapper acquires a single reader against the inner stream and pulls chunk-by-chunk inside the outer stream's `pull` callback. This preserves backpressure end-to-end: the outer consumer's pull cadence drives the inner sub-pipeline's read cadence.
+- On error, the wrapper sets a `closed` flag, closes the outer controller, and calls `onError`. Subsequent pulls return `done` without re-invoking the inner reader.
+- Cancellation of the wrapper propagates to the inner reader via `reader.cancel(reason)`. Cancellation errors from the inner stream are swallowed — they are part of the isolated branch by definition.
+- The error code is `STREAM.BRANCH_FAILED` (registered in `errors/stream-error.ts`). Inspect `err.code` to detect a branch failure programmatically; inspect `err.cause` to recover the original error.
+- `safeBranch` does not retry or restart the inner pipeline. After failure, the wrapper is terminal. To retry, pair `safeBranch` with `@suaveplan/retry` upstream of the factory.

docs/foundation/flow/streams/src/pipeline/zip/zip.md

@@ -0,0 +1,107 @@
+# zip
+
+<!-- begin:auto-docs -->
+_(no submodules)_
+<!-- end:auto-docs -->
+## Purpose
+
+`zip` performs a sample-aligned merge of N input `ReadableStream`s into a single
+stream of tuples. Unlike `merge`, which interleaves values in arrival order,
+`zip` waits for every input to produce a chunk before emitting one tuple per
+round. This makes it the canonical primitive for combining parallel time-series,
+correlated event streams, or any pair of producers whose values must be paired
+position-by-position rather than raced.
+
+The operation is fully back-pressured — readers are pulled exactly once per
+output tuple — and is preserved end-to-end by propagating `cancel()` from the
+output stream to every input.
+
+## Features
+
+- Tuple output with full TypeScript inference: `zip([s1, s2] as const)` is
+    typed as `ReadableStream<[T1, T2]>`.
+- Three short-input policies: `close` (default), `error`, `pad`.
+- `StreamZipLengthMismatchError` (code `STREAM.ZIP_LENGTH_MISMATCH`) under
+    `error` policy carries the index of the input that ended early.
+- `pad` policy substitutes the configured `padValue` for any input that has
+    already ended, continuing until the longest input closes.
+- Cancellation propagates to every input reader.
+- Empty input list returns an immediately-closed stream.
+
+## Usage: Basic Example
+
+```ts
+import { fromArray, toArray, zip } from "@suaveplan/streams";
+
+const numbers = fromArray([1, 2, 3]);
+const letters = fromArray(["a", "b", "c"]);
+
+const tuples = await toArray(zip([numbers, letters] as const));
+// tuples === [[1, "a"], [2, "b"], [3, "c"]]
+```
+
+## Usage: Advanced Example
+
+```ts
+import { fromArray, toArray, zip } from "@suaveplan/streams";
+import { StreamZipLengthMismatchError } from "@suaveplan/streams/errors/stream-error";
+
+// Pad short inputs so every tuple has both slots filled.
+const padded = await toArray(
+    zip(
+        [fromArray([1, 2, 3, 4]), fromArray(["a", "b"])] as const,
+        { onShortInput: "pad", padValue: null }
+    )
+);
+// padded === [[1, "a"], [2, "b"], [3, null], [4, null]]
+
+// Strict mode — any length mismatch is a hard error.
+try {
+    await toArray(
+        zip(
+            [fromArray([1, 2, 3]), fromArray(["a", "b"])] as const,
+            { onShortInput: "error" }
+        )
+    );
+} catch (err) {
+    if (err instanceof StreamZipLengthMismatchError) {
+        // err.code === "STREAM.ZIP_LENGTH_MISMATCH"
+    }
+}
+```
+
+## API Reference
+
+### `zip<T extends readonly unknown[]>(streams, options?): ReadableStream<T>`
+
+- `streams: { readonly [K in keyof T]: ReadableStream<T[K]> }` — tuple of input
+    streams. Pass with `as const` for tuple inference.
+- `options.onShortInput?: 'close' | 'error' | 'pad'` — short-input behaviour.
+    Defaults to `'close'`.
+- `options.padValue?: unknown` — value substituted for ended inputs when
+    `onShortInput === 'pad'`. Defaults to `undefined`.
+- **Returns:** `ReadableStream<T>` emitting one tuple per round.
+- **Throws:** the output errors with `StreamZipLengthMismatchError` when
+    `onShortInput === 'error'` and any input ends before the others.
+
+### `ZipOptions<TValue = unknown>`
+
+Options bag passed to `zip`. See above for fields.
+
+### `ZipShortInputPolicy`
+
+String-literal union: `'close' | 'error' | 'pad'`.
+
+## Implementation Notes
+
+- Each pull round issues `read()` on every still-open reader in parallel via
+    `Promise.all`. This keeps tuple emission strictly ordered while letting the
+    runtime pipeline reads.
+- The `pad` policy tracks per-input "ended" flags so closed readers are not
+    re-read on subsequent rounds — only `padValue` is substituted in their slot.
+- `cancel()` on the output settles cancels on every input via
+    `Promise.allSettled`, so a single failing cancel cannot leak others.
+- Read errors propagate verbatim via `controller.error(err)`; only the
+    `'error'` short-input policy wraps in `StreamZipLengthMismatchError`.
+- An empty input list short-circuits to an immediately-closed stream rather
+    than starting any readers.