diff --git a/doc/api/webstreams.md b/doc/api/webstreams.md
index cf03323913b07a..c53c99a32b0e98 100644
--- a/doc/api/webstreams.md
+++ b/doc/api/webstreams.md
@@ -1606,6 +1606,7 @@ import {
   buffer,
   json,
   text,
+  concat,
 } from 'node:stream/consumers';
 ```
 
@@ -1616,6 +1617,7 @@ const {
   buffer,
   json,
   text,
+  concat,
 } = require('node:stream/consumers');
 ```
 
@@ -1846,14 +1848,195 @@ text(readable).then((data) => {
 });
 ```
 
+#### `streamConsumers.concat(sources[, options])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* `sources` {Iterable|AsyncIterable} An iterable or async iterable of sources.
+  Each item may be a {Stream}, a {ReadableStream}, an {Iterable} or
+  {AsyncIterable} of chunks, a chunk ({string}, {Buffer}, {TypedArray} or
+  {DataView}), or a {Function} returning any of those, optionally as a
+  {Promise}.
+* `options` {Object}
+  * `signal` {AbortSignal} Aborting the signal fails the returned iterator
+    with `signal.reason` and destroys every source.
+  * `lazy` {boolean} Whether sources are obtained one at a time rather than up
+    front. **Default:** `true` when `sources` is an async iterable, `false`
+    otherwise. See [Collections and producers][].
+  * `destroyOnReturn` {boolean} When `false`, sources that were never consumed
+    are returned to the caller live and without error handlers attached rather
+    than being destroyed. **Default:** `true`.
+* Returns: {AsyncIterator} Yields the chunks of each source in turn.
+
+Reads a sequence of sources as though they were one. Each source is consumed
+to completion before the next is started.
+
+Unlike the other utility consumers, `concat()` does not consume anything by
+itself and does not return a promise. It returns an async iterator, which the
+other consumers in this module accept directly.
+
+```mjs
+import { concat, text } from 'node:stream/consumers';
+import { createReadStream } from 'node:fs';
+
+const document = await text(concat([
+  createReadStream('header.txt'),
+  createReadStream('body.txt'),
+  createReadStream('footer.txt'),
+]));
+```
+
+```cjs
+const { concat, text } = require('node:stream/consumers');
+const { createReadStream } = require('node:fs');
+
+text(concat([
+  createReadStream('header.txt'),
+  createReadStream('body.txt'),
+  createReadStream('footer.txt'),
+])).then((document) => {
+  console.log(document.length);
+});
+```
+
+Returning an iterator rather than a stream means `concat()` has no
+`objectMode` of its own to configure and no `highWaterMark`: chunks pass
+through exactly as their source produced them, and the sequence is read only
+as fast as it is consumed. Use [`stream.Readable.from()`][] to obtain a
+stream:
+
+```mjs
+import { concat } from 'node:stream/consumers';
+import { Readable } from 'node:stream';
+
+const readable = Readable.from(concat(sources), { objectMode: false });
+```
+
+`sources` must be a collection, not a single source. Passing a string or a
+{Buffer} throws [`ERR_INVALID_ARG_TYPE`][], since iterating one would yield
+its individual characters or bytes rather than treating it as one chunk. Wrap
+it in an array instead.
+
+##### Collections and producers
+
+A {stream.Readable} begins consuming its underlying resource as soon as it is
+constructed, and one that emits `'error'` with no listener attached will throw
+an uncaught exception. `concat()` therefore attaches an error handler to every
+source at the moment it obtains a reference to it, and never holds a reference
+it is not listening to.
+
+The shape of `sources` determines when those references are obtained:
+
+* When `sources` is a synchronous iterable it is treated as a **collection**.
+  It is drained when `concat()` is called and every stream it contains is
+  guarded immediately. Synchronous iteration cannot block, and anything
+  already in the collection is already running, so nothing is gained by
+  deferring.
+
+* When `sources` is an async iterable it is treated as a **producer**. Sources
+  are obtained one at a time and guarded as they arrive; the producer is never
+  read ahead of the consumer, and an unbounded producer is safe.
+
+There is no way to distinguish a collection that already holds live streams
+from a producer that creates one per iteration — a `Set` of open file streams
+and a generator that opens one per `yield` present the identical interface —
+so the two cases are separated by the protocol rather than by inspection. Use
+`options.lazy` where the default is wrong, for example with a synchronous
+generator that opens a resource per iteration:
+
+```mjs
+import { concat } from 'node:stream/consumers';
+import { createReadStream } from 'node:fs';
+
+function* openEach(paths) {
+  for (const path of paths) yield createReadStream(path);
+}
+
+// Without `lazy`, every file would be opened when concat() is called.
+const parts = concat(openEach(paths), { lazy: true });
+```
+
+Alternatively, supply factories. A factory is inert until reached, which gives
+laziness under either default and never leaves an unguarded source:
+
+```mjs
+import { concat } from 'node:stream/consumers';
+import { createReadStream } from 'node:fs';
+
+const parts = concat(paths.map((path) => () => createReadStream(path)));
+```
+
+Guarding applies only to {stream.Readable} sources. A {ReadableStream} or an
+async iterable stores its own failure and surfaces it on the next read, and so
+has no unhandled-`'error'` path to protect against.
+
+##### Error handling
+
+If any source fails — including one that has not been read yet — iteration
+rejects with that error. The first error wins; later ones are discarded. The
+error is raised as soon as it occurs rather than when the current source
+finishes, so a failure elsewhere in the sequence will not be delayed behind a
+source that is slow or stalled.
+
+`concat()` calls `stream.destroy(err)` on every stream it is still holding,
+and cancels every {ReadableStream}, when iteration ends for any reason other
+than reading every source to completion: an error, an abort, or leaving a
+`for await...of` loop through `break`, `return` or `throw`. Sources that were
+read to completion are left as their own semantics leave them, with the
+listeners `concat()` added removed.
+
+Set `destroyOnReturn` to `false` to take ownership of the unconsumed sources
+instead. They are handed back live and unguarded, and an `'error'` from one is
+then the caller's to handle.
+
+```mjs
+import { concat } from 'node:stream/consumers';
+import process from 'node:process';
+
+const ac = new AbortController();
+setTimeout(() => ac.abort(), 5000);
+
+try {
+  for await (const chunk of concat(sources, { signal: ac.signal })) {
+    process.stdout.write(chunk);
+  }
+} catch (err) {
+  if (err.name === 'AbortError') {
+    // Every source has been destroyed.
+  } else {
+    throw err;
+  }
+}
+```
+
+A source that never produces data and never ends cannot be escaped with
+`break` or by calling `return()` on the iterator: an async iterator suspended
+awaiting a read cannot be interrupted, and the request queues behind the read
+that will not complete. `options.signal` is the only way to abandon a stalled
+source.
+
+`options.signal` aborts the iteration and destroys the sources. It is not
+forwarded to the sources themselves, so a source with its own `signal` support
+is destroyed rather than cancelled at the underlying operation. Pass the same
+signal to the source directly, or use [`stream.addAbortSignal()`][], where
+that distinction matters.
+
+[Collections and producers]: #collections-and-producers
 [Streams]: stream.md
 [WHATWG Streams Standard]: https://streams.spec.whatwg.org/
+[`ERR_INVALID_ARG_TYPE`]: errors.md#err_invalid_arg_type
 [`stream.Duplex.fromWeb`]: stream.md#streamduplexfromwebpair-options
 [`stream.Duplex.toWeb`]: stream.md#streamduplextowebstreamduplex-options
 [`stream.Duplex`]: stream.md#class-streamduplex
+[`stream.Readable.from()`]: stream.md#streamreadablefromiterable-options
 [`stream.Readable.fromWeb`]: stream.md#streamreadablefromwebreadablestream-options
 [`stream.Readable.toWeb`]: stream.md#streamreadabletowebstreamreadable-options
 [`stream.Readable`]: stream.md#class-streamreadable
 [`stream.Writable.fromWeb`]: stream.md#streamwritablefromwebwritablestream-options
 [`stream.Writable.toWeb`]: stream.md#streamwritabletowebstreamwritable
 [`stream.Writable`]: stream.md#class-streamwritable
+[`stream.addAbortSignal()`]: stream.md#streamaddabortsignalsignal-stream
diff --git a/lib/stream/consumers.js b/lib/stream/consumers.js
index a0d4a5d541ad66..7c50ed1dcb07c4 100644
--- a/lib/stream/consumers.js
+++ b/lib/stream/consumers.js
@@ -94,4 +94,5 @@ module.exports = {
   bytes,
   text,
   json,
+  concat: require('stream/consumers/concat'),
 };
diff --git a/lib/stream/consumers/concat.js b/lib/stream/consumers/concat.js
new file mode 100644
index 00000000000000..ebb076bad9e42a
--- /dev/null
+++ b/lib/stream/consumers/concat.js
@@ -0,0 +1,398 @@
+'use strict';
+
+/**
+ * concat(sources[, options]) - lazily concatenate readable sources into a
+ * single async iterator.
+ *
+ * `sources` is one iterable or async iterable of sources. Each item may be:
+ *   - stream.Readable (including legacy "streams1" objects)
+ *   - web ReadableStream
+ *   - a sync/async iterable of chunks
+ *   - a chunk (string, Buffer, TypedArray) - yielded as-is
+ *   - a factory: () => any of the above, possibly async, built on demand
+ *
+ * Returns a plain async iterator. It deliberately does not return a stream, so
+ * it never has to guess an objectMode. Compose to taste:
+ *
+ *   Readable.from(concat(sources))
+ *   await text(concat(sources))
+ *
+ * -- Signature ----------------------------------------------------------------
+ *
+ * One collection, one options bag. Not variadic. Sources usually arrive from a
+ * calling context already in a collection, so varargs would force a spread at
+ * every call site, and - more importantly - it would permanently occupy the
+ * second position. `pipeline()` went variadic and has been paying for it ever
+ * since: the trailing argument had to be special-cased as the callback, then
+ * `{ signal }` had to squeeze in ahead of it. `Buffer.concat(list, ...)` and
+ * `Readable.from(iterable, options)` are the shapes to copy.
+ *
+ * -- Eagerness contract -------------------------------------------------------
+ *
+ * A Node stream starts working the moment it is constructed, and an 'error' it
+ * emits with no listener attached takes down the process. So the invariant is:
+ *
+ *     never hold a reference to a live Node stream we are not listening to.
+ *
+ * We cannot ask an iterable whether its items already exist. A Set of open file
+ * streams and a generator that opens one per pull have identical interfaces.
+ * So the input protocol picks the default:
+ *
+ *   sync iterable  -> collection semantics. Drained at call time; every live
+ *                     Node stream in it is guarded immediately. Sync iteration
+ *                     cannot block, and anything already in there is already
+ *                     running, so there is nothing to lose by looking.
+ *
+ *   async iterable -> producer semantics. Pulled one at a time, guarded at the
+ *                     instant of acquisition, never read ahead. Choosing an
+ *                     async producer is how you declare that pulling costs
+ *                     something (or that the sequence may be unbounded).
+ *
+ *   function        -> laziness, under either default. A factory is inert, so a
+ *                      sync collection of factories can be drained safely.
+ *
+ * ...and `options.lazy` overrides it when the default guesses wrong - a sync
+ * generator that opens a stream per pull wants `{ lazy: true }`.
+ *
+ * The guarding only applies to Node streams. Web streams and async iterables
+ * store their failure and hand it over on the next read; they have no
+ * unhandled-'error' failure mode and need no babysitting.
+ *
+ * -- Options ------------------------------------------------------------------
+ *
+ *   signal           AbortSignal. Aborting fails the iterator with
+ *                    `signal.reason` and tears down every source.
+ *
+ *   lazy             Override the sync/async default above. `true` pulls one
+ *                    source at a time; `false` drains `sources` up front and
+ *                    guards everything. Draining an unbounded async producer
+ *                    will hang, so `false` is only for bounded ones.
+ *
+ *   destroyOnReturn  Default true. When false, sources are handed back live and
+ *                    unguarded on early return or failure - you own them, and
+ *                    an unhandled 'error' from one is yours to catch.
+ */
+
+const {
+  ArrayBuffer,
+  ArrayBufferIsView,
+  ArrayFrom,
+  Boolean,
+  Promise,
+  PromisePrototypeThen,
+  PromiseResolve,
+  SafeMap,
+  SafePromiseRace,
+  SymbolAsyncIterator,
+  SymbolIterator,
+} = primordials;
+
+const {
+  codes: {
+    ERR_INVALID_ARG_TYPE,
+  },
+} = require('internal/errors');
+
+const {
+  validateObject,
+  validateAbortSignal,
+} = require('internal/validators');
+
+const Readable = require('internal/streams/readable');
+
+
+const isFn = (v) => typeof v === 'function';
+
+const noop = () => {};
+
+// Release an iterator without waiting on it. A stream async iterator's return()
+// does not settle until any outstanding next() does, and the next() abandoned
+// when racing the latch never will - so awaiting this deadlocks against a
+// source that has stalled. Destroying the stream is what settles that pending
+// read, and teardown() has already done it by the time this runs.
+function closeQuietly(iterator) {
+  if (!iterator || !isFn(iterator.return)) return;
+  try {
+    const result = iterator.return();
+    if (result !== null && typeof result === 'object' && isFn(result.then)) {
+      PromisePrototypeThen(PromiseResolve(result), noop, noop);
+    }
+  } catch {
+    // Already gone.
+  }
+}
+
+const isNodeStream = (o) =>
+  o !== null && typeof o === 'object' && isFn(o.pipe) && isFn(o.on);
+
+const isWebReadable = (o) =>
+  o !== null && typeof o === 'object' && isFn(o.getReader) && isFn(o.cancel) &&
+  !isNodeStream(o);
+
+const isAsyncIterable = (o) =>
+  o !== null && o !== undefined && isFn(o[SymbolAsyncIterator]);
+
+const isSyncIterable = (o) =>
+  o !== null && o !== undefined && isFn(o[SymbolIterator]);
+
+// Valid as an item, never as the `sources` collection, so concat('abc') is a
+// mistake rather than three single-character sources.
+const isChunk = (o) =>
+  typeof o === 'string' || ArrayBufferIsView(o) || o instanceof ArrayBuffer;
+
+// ---------------------------------------------------------------------------
+// source normalization
+// ---------------------------------------------------------------------------
+
+async function* fromWebReadable(rs) {
+  const reader = rs.getReader();
+  let drained = false;
+  try {
+    for (;;) {
+      const { done, value } = await reader.read();
+      if (done) {
+        drained = true;
+        return;
+      }
+      yield value;
+    }
+  } finally {
+    if (!drained) {
+      reader.cancel().catch(() => {}); // Best effort: tell the producer we left
+    }
+    reader.releaseLock();
+  }
+}
+
+function toAsyncIterator(source) {
+  if (isChunk(source)) {
+    return (async function* () { yield source; })();
+  }
+  if (isNodeStream(source)) {
+    // Legacy streams1 have no Symbol.asyncIterator. This is the case ronag
+    // flagged on the original PR; pipeline() solves it the same way.
+    const modern = isAsyncIterable(source) ? source :
+      new Readable({ objectMode: true }).wrap(source);
+    return modern[SymbolAsyncIterator]();
+  }
+  if (isWebReadable(source)) {
+    return isAsyncIterable(source) ? source[SymbolAsyncIterator]() :
+      fromWebReadable(source);
+  }
+  if (isAsyncIterable(source)) return source[SymbolAsyncIterator]();
+  if (isSyncIterable(source)) {
+    return (async function* () { yield* source; })();
+  }
+  throw new ERR_INVALID_ARG_TYPE('source',
+                                 ['Stream', 'Iterable', 'AsyncIterable'],
+                                 source);
+}
+
+// ---------------------------------------------------------------------------
+// error latch
+// ---------------------------------------------------------------------------
+
+function createLatch(signal) {
+  let reject;
+  const trigger = new Promise((_, rej) => { reject = rej; });
+  trigger.catch(() => {}); // Never an unhandled rejection; we always race it
+
+  const guards = new SafeMap(); // stream -> remove its 'error' listener
+  const state = { error: null, done: false };
+  let detachSignal = () => {};
+
+  const fail = (err) => {
+    if (state.error === null && !state.done) {
+      state.error = err;
+      reject(err);
+    }
+  };
+
+  if (signal) {
+    if (signal.aborted) {
+      fail(signal.reason);
+    } else {
+      const onAbort = () => fail(signal.reason);
+      signal.addEventListener('abort', onAbort, { once: true });
+      detachSignal = () => signal.removeEventListener('abort', onAbort);
+    }
+  }
+
+  return {
+    trigger,
+    get error() { return state.error; },
+
+    // One 'error' listener goes on at the instant we take a reference to a
+    // stream. It is both the floor - while it is attached the stream cannot
+    // throw an unhandled 'error' at the process, including when *we* destroy
+    // it during teardown - and the observer that reports a sibling's failure.
+    //
+    // This deliberately does not use end-of-stream. eos() reports end, close
+    // and premature-close as well, none of which we need: we know a source is
+    // finished because our own loop drained it. All we want is the error, and
+    // a listener cannot hand back the wrong kind of cleanup value.
+    track(source) {
+      if (!isNodeStream(source) || guards.has(source)) return;
+      const onError = (err) => fail(err);
+      source.on('error', onError);
+      guards.set(source, () => source.removeListener('error', onError));
+
+      // A source that failed before we ever saw it emitted 'error' with no
+      // listener attached; `errored` is how that stays observable afterwards.
+      if (source.errored) fail(source.errored);
+    },
+
+    // Read to completion, no failure. Hand the stream back exactly as we found
+    // it. Membership is decided by presence in the map, never by the
+    // truthiness of what is stored against it.
+    release(source) {
+      if (!guards.has(source)) return;
+      guards.get(source)();
+      guards.delete(source);
+    },
+
+    fail,
+
+    // Destroy with the listeners still attached: destroy(err) emits 'error'
+    // asynchronously, and by delivery time anything detached first would be
+    // gone. state.done already blocks fail(), so the echo is absorbed
+    // silently. Only the hand-back path detaches, because there the caller is
+    // taking ownership.
+    //
+    // forEach hands the value first and the key second, so there is no
+    // .values()/.keys() call to lose in translation and no iterator protocol
+    // involved.
+    teardown(err, destroy) {
+      state.done = true;
+      detachSignal();
+
+      const pairs = [];
+      guards.forEach((detach, source) => {
+        pairs.push(source, detach);
+      });
+      guards.clear();
+
+      for (let i = 0; i < pairs.length; i += 2) {
+        const source = pairs[i];
+        if (!destroy) {
+          pairs[i + 1](); // Caller owns it now
+        } else if (isFn(source.destroy) && !source.destroyed) {
+          source.destroy(err ?? undefined);
+        }
+      }
+    },
+  };
+}
+
+// ---------------------------------------------------------------------------
+// concat
+// ---------------------------------------------------------------------------
+
+function concat(sources, options = undefined) {
+
+  if (isChunk(sources)) {
+    throw new ERR_INVALID_ARG_TYPE('sources', ['Iterable', 'AsyncIterable'], sources);
+  }
+  const isAsync = isAsyncIterable(sources);
+  if (!isAsync && !isSyncIterable(sources)) {
+    throw new ERR_INVALID_ARG_TYPE('sources', ['Iterable', 'AsyncIterable'], sources);
+  }
+
+  // Let the validators decide. Pre-screening with `typeof === 'object'` lets an
+  // array or a stream through as an options bag, which validateObject rejects.
+  if (options !== undefined) validateObject(options, 'options');
+  const { signal, lazy, destroyOnReturn = true } = options ?? {};
+  if (signal !== undefined && signal !== null) {
+    validateAbortSignal(signal, 'options.signal');
+  }
+
+  const latch = createLatch(signal);
+  const deferred = lazy === undefined ? isAsync : Boolean(lazy);
+
+  // Guard at call time, not on first next(). A collection handed to concat()
+  // holds running streams whether or not anyone ever iterates the result.
+  // (An async collection cannot be drained synchronously, so an eager pass
+  // over one happens on first pull instead - nothing in it is reachable, and
+  // therefore nothing in it is at risk, until then.)
+  let entries = sources;
+  if (!deferred && !isAsync) {
+    entries = ArrayFrom(sources);
+    for (const entry of entries) {
+      if (!isFn(entry)) latch.track(entry);
+    }
+  }
+
+  return run(entries, deferred, isAsync, latch, destroyOnReturn);
+}
+
+async function* run(entries, deferred, isAsync, latch, destroyOnReturn) {
+  let outer = null;
+  let inner = null;
+
+  try {
+    if (!deferred && isAsync) {
+      // Bounded async producer, eagerness explicitly requested: drain first,
+      // then guard everything before reading a single byte.
+      const drained = [];
+      for await (const entry of entries) drained.push(entry);
+      for (const entry of drained) {
+        if (!isFn(entry)) latch.track(entry);
+      }
+      if (latch.error) throw latch.error;
+      entries = drained;
+      isAsync = false;
+    }
+
+    const outerIsAsync = deferred && isAsync;
+    outer = outerIsAsync ? entries[SymbolAsyncIterator]() : entries[SymbolIterator]();
+
+    for (;;) {
+      // A sync iterator's next() returns a plain result object, not a promise,
+      // and there is nothing to race anyway: a synchronous pull cannot be
+      // outrun by the latch. Check the latch directly instead.
+      const step = outerIsAsync ?
+        await SafePromiseRace([PromiseResolve(outer.next()), latch.trigger]) :
+        outer.next();
+      if (latch.error) throw latch.error;
+      if (step.done) break;
+
+      let source = step.value;
+
+      if (isFn(source)) {
+        // Factory. Nothing existed until now, so nothing could have failed.
+        source = await SafePromiseRace([
+          (async () => source())(),
+          latch.trigger,
+        ]);
+        latch.track(source);
+      } else if (deferred) {
+        latch.track(source);
+      }
+
+      inner = toAsyncIterator(source);
+
+      for (;;) {
+        // Race the read against a sibling failing, so we abort immediately
+        // rather than waiting on a chunk that may never arrive.
+        const chunk = await SafePromiseRace([PromiseResolve(inner.next()), latch.trigger]);
+        if (chunk.done) break;
+        yield chunk.value;
+      }
+
+      inner = null;
+      latch.release(source);
+    }
+  } catch (err) {
+    latch.fail(err);
+    throw latch.error ?? err;
+  } finally {
+    // Destroy before releasing the iterators, not after: destroying is what
+    // settles the read abandoned when the latch fired, and a stream iterator's
+    // return() will not complete until it does.
+    latch.teardown(latch.error, destroyOnReturn);
+    closeQuietly(inner);
+    closeQuietly(outer);
+  }
+}
+
+module.exports = concat;
diff --git a/test/parallel/test-stream-consumers-concat.js b/test/parallel/test-stream-consumers-concat.js
new file mode 100644
index 00000000000000..0958ad3412bbfa
--- /dev/null
+++ b/test/parallel/test-stream-consumers-concat.js
@@ -0,0 +1,513 @@
+'use strict';
+
+require('../common');
+
+const assert = require('node:assert');
+
+const {
+  EventEmitter
+} = require('node:events');
+
+const {
+  Readable
+} = require('node:stream');
+
+const {
+  concat,
+  text
+} = require('node:stream/consumers');
+
+const {
+  test,
+} = require('node:test');
+
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
+
+
+const collect = async (it) => {
+  const out = [];
+  for await (const chunk of it) out.push(String(chunk));
+  return out;
+};
+
+// A stream that yields its chunks slowly, so siblings have time to misbehave.
+const slow = (chunks, delay = 10) =>
+  Readable.from((async function* () {
+    for (const c of chunks) {
+      await sleep(delay);
+      yield c;
+    }
+  })());
+
+// A live, idle stream that blows up on a timer with nobody reading it.
+const bombs = (ms, err = new Error('boom')) => {
+  const s = new Readable({ read() {} });
+  setTimeout(() => s.destroy(err), ms);
+  return s;
+};
+
+// ---------------------------------------------------------------------------
+// signature
+// ---------------------------------------------------------------------------
+
+
+test('rejects a chunk as the collection', () => {
+  assert.throws(() => concat('abc'), /must be an instance of Iterable or AsyncIterable/);
+  assert.throws(() => concat(Buffer.from('abc')), /must be an instance of Iterable or AsyncIterable/);
+});
+
+test('rejects an invalid source inside the collection', async () => {
+  await assert.rejects(
+    collect(concat([Readable.from(['a']), 42])),
+    /argument must be an instance of Stream, Iterable, or AsyncIterable/);
+});
+
+test('rejects a non-iterable collection', () => {
+  assert.throws(() => concat(42), /must be an instance of Iterable or AsyncIterable/);
+  assert.throws(() => concat(null), /must be an instance of Iterable or AsyncIterable/);
+});
+
+test('rejects a non-object options bag', () => {
+  assert.throws(() => concat([], 'nope'), /argument must be of type object/);
+  assert.throws(() => concat([], { signal: 'nope' }), /must be an instance of AbortSignal/);
+});
+
+// ---------------------------------------------------------------------------
+// invariants
+//
+// These assert the guarding contract directly and use no timers. When the
+// timing-dependent tests below them fail, these say whether the listeners were
+// ever attached at all - the difference between a broken guard and a broken
+// assertion.
+// ---------------------------------------------------------------------------
+
+test('every source is guarded before the first read', async () => {
+  const sources = [
+    new Readable({ read() {} }),
+    new Readable({ read() {} }),
+    new Readable({ read() {} }),
+  ];
+
+  const it = concat(sources);
+  for (let i = 0; i < sources.length; i++) {
+    assert.ok(sources[i].listenerCount('error') > 0,
+              `sources[${i}] left unguarded at call time`);
+  }
+  await it.return();
+});
+
+test('a source read to completion keeps none of our listeners', async () => {
+  // Node's own async iterator installs a permanent 'error' handler on any
+  // stream it consumes, so the count never reaches zero. Measure against a
+  // stream consumed without concat() to isolate the listeners we added.
+  const control = Readable.from(['x']);
+  for await (const chunk of control) assert.strictEqual(String(chunk), 'x');
+  const baseline = control.listenerCount('error');
+
+  const a = Readable.from(['a']);
+  const b = Readable.from(['b']);
+  assert.deepStrictEqual(await collect(concat([a, b])), ['a', 'b']);
+
+  assert.strictEqual(a.listenerCount('error'), baseline); // Listeners left on a
+  assert.strictEqual(b.listenerCount('error'), baseline); // Listeners left on b
+});
+
+test('a guarded source survives an error with nobody reading it', async () => {
+  const idle = new Readable({ read() {} });
+  const it = concat([new Readable({ read() {} }), idle]);
+
+  // Synchronous destroy: an unguarded source throws here rather than as an
+  // uncaught exception on a later tick, which is far easier to attribute.
+  idle.destroy(new Error('guarded'));
+  await assert.rejects(collect(it), /guarded/);
+});
+
+test('a source that failed before concat() saw it still reports', async () => {
+  const dead = new Readable({ read() {} });
+  dead.on('error', () => {});
+  dead.destroy(new Error('already gone'));
+  await sleep(5); // Let the 'error' event fire before concat() ever sees it
+
+  await assert.rejects(
+    collect(concat([Readable.from(['a']), dead])), /already gone/);
+});
+
+// ---------------------------------------------------------------------------
+// basics
+// ---------------------------------------------------------------------------
+
+test('concatenates in order', async () => {
+  const out = await collect(concat([
+    Readable.from(['a', 'b']),
+    Readable.from(['c']),
+    Readable.from(['d', 'e']),
+  ]));
+  assert.deepStrictEqual(out, ['a', 'b', 'c', 'd', 'e']);
+});
+
+test('mixes streams, iterables and chunks', async () => {
+  const out = await collect(concat([
+    Readable.from(['a']),
+    ['b', 'c'],
+    'd',
+    (function* () { yield 'e'; })(),
+  ]));
+  assert.deepStrictEqual(out, ['a', 'b', 'c', 'd', 'e']);
+});
+
+// The case mcollina raised on the PR: an unread sibling errors while we are
+// still draining an earlier stream. Naively this is an unhandled 'error' event
+// and the process dies. Here it must surface as a normal rejection.
+test('an idle sibling erroring does not crash the process', async () => {
+  const it = concat([slow(['a', 'b', 'c', 'd'], 20), bombs(15), Readable.from(['z'])]);
+  await assert.rejects(collect(it), /boom/);
+});
+
+test('the failure aborts promptly, without draining the current stream', async () => {
+  const started = Date.now();
+  // Ten chunks at 20ms would take 200ms; the bomb goes off at 15ms.
+  const it = concat([slow('0123456789'.split(''), 20), bombs(15)]);
+
+  await assert.rejects(collect(it), /boom/);
+  assert.ok(Date.now() - started < 120, `aborted late: ${Date.now() - started}ms`);
+});
+
+test('a failure destroys every other source', async () => {
+  const a = slow(['a', 'b', 'c'], 20);
+  const c = new Readable({ read() {} });
+
+  await assert.rejects(collect(concat([a, bombs(15), c])), /boom/);
+  await sleep(10);
+  assert.ok(a.destroyed, 'current source destroyed');
+  assert.ok(c.destroyed, 'queued source destroyed');
+});
+
+test('chunks yielded before a failure are not withdrawn', async () => {
+  // Fail an unread sibling at a known point in the output rather than on a
+  // timer, so the prefix is exact rather than a race between two delays.
+  const bomb = new Readable({ read() {} });
+  const it = concat([
+    Readable.from(['a', 'b']),
+    slow(['c', 'd'], 100), // Reached, but never gets to deliver
+    Readable.from(['e']),  // Never reached
+    bomb,
+  ]);
+
+  const captured = [];
+  let error = null;
+
+  try {
+    for await (const chunk of it) {
+      captured.push(String(chunk));
+      if (captured.length === 2) bomb.destroy(new Error('boom'));
+    }
+  } catch (err) {
+    error = err;
+  }
+
+  assert.ok(error, 'iteration should have failed');
+  assert.strictEqual(error.message, 'boom');
+  // Everything delivered before the failure is kept, in order, and nothing
+  // from a later source leaks out after it.
+  assert.deepStrictEqual(captured, ['a', 'b']);
+});
+
+test('a failure mid-source keeps the chunks already taken from it', async () => {
+  const bomb = new Readable({ read() {} });
+  const it = concat([
+    Readable.from(['a']),
+    slow(['b', 'c', 'd'], 100),
+    bomb,
+  ]);
+
+  const captured = [];
+  let error = null;
+
+  try {
+    for await (const chunk of it) {
+      captured.push(String(chunk));
+      // Fires partway through the second source, not between sources.
+      if (captured.length === 2) bomb.destroy(new Error('boom'));
+    }
+  } catch (err) {
+    error = err;
+  }
+
+  assert.ok(error, 'iteration should have failed');
+  assert.strictEqual(error.message, 'boom');
+  assert.deepStrictEqual(captured, ['a', 'b']);
+});
+
+test('a source is never pulled ahead of what was delivered', async () => {
+  // Recovery depends on this. A caller that wraps a source to record its
+  // progress must never record a chunk the consumer did not receive, or it
+  // will resume from the wrong offset. concat() pulls one chunk at a time and
+  // only on demand, so the wrapper cannot run ahead of the consumer.
+  const state = { delivered: 0 };
+  const bomb = new Readable({ read() {} });
+
+  async function* tracked() {
+    for (const c of ['aa', 'bb', 'cc', 'dd']) {
+      await sleep(20);
+      yield c;
+      state.delivered += c.length; // Reached only once the consumer took it
+    }
+  };
+
+  setTimeout(() => bomb.destroy(new Error('boom')), 50);
+
+  const received = [];
+  await assert.rejects(async () => {
+    for await (const chunk of concat([tracked(), bomb])) {
+      received.push(String(chunk));
+    }
+  }, /boom/);
+
+  assert.strictEqual(state.delivered, received.join('').length);
+});
+
+test('the first error wins', async () => {
+  const it = concat([
+    slow(['a', 'b', 'c'], 30),
+    bombs(10, new Error('first')),
+    bombs(20, new Error('second')),
+  ]);
+  await assert.rejects(collect(it), /first/);
+});
+
+test('breaking early destroys unconsumed sources', async () => {
+  const a = Readable.from(['a', 'b']);
+  const b = new Readable({ read() {} });
+  const c = new Readable({ read() {} });
+
+  for await (const chunk of concat([a, b, c])) {
+    assert.strictEqual(String(chunk), 'a');
+    break;
+  }
+  await sleep(10);
+  assert.ok(a.destroyed && b.destroyed && c.destroyed);
+});
+
+test('an error inside the current source propagates', async () => {
+  const bad = Readable.from((async function* () {
+    yield 'a';
+    throw new Error('inner');
+  })());
+  const next = new Readable({ read() {} });
+
+  await assert.rejects(collect(concat([bad, next])), /inner/);
+  await sleep(10);
+  assert.ok(next.destroyed, 'queued source destroyed');
+});
+
+// ---------------------------------------------------------------------------
+// options — the slot the variadic signature would have cost us
+// ---------------------------------------------------------------------------
+
+test('signal: aborting mid-read fails the iterator', async () => {
+  const ac = new AbortController();
+  const b = new Readable({ read() {} });
+  setTimeout(() => ac.abort(), 15);
+
+  await assert.rejects(
+    collect(concat([slow(['a', 'b', 'c'], 25), b], { signal: ac.signal })),
+    (err) => err.name === 'AbortError');
+  await sleep(10);
+  assert.ok(b.destroyed, 'queued source destroyed on abort');
+});
+
+test('aborting is the only way out of a stalled source', async () => {
+  // A source that never produces and never ends cannot be escaped with break
+  // or return(): an async generator suspended at an await cannot be
+  // force-returned, the request just queues behind the pending read. The
+  // signal is the escape hatch - and teardown must destroy before releasing
+  // the iterators, or it deadlocks on the read it abandoned.
+  const ac = new AbortController();
+  const stalled = new Readable({ read() {} });
+  const queued = new Readable({ read() {} });
+  setTimeout(() => ac.abort(), 15);
+
+  await assert.rejects(
+    collect(concat([stalled, queued], { signal: ac.signal })),
+    (err) => err.name === 'AbortError');
+
+  await sleep(10);
+  assert.ok(stalled.destroyed, 'stalled source destroyed');
+  assert.ok(queued.destroyed, 'queued source destroyed');
+});
+
+test('signal: an already-aborted signal fails on first pull', async () => {
+  const signal = AbortSignal.abort(new Error('too late'));
+  await assert.rejects(collect(concat([Readable.from(['a'])], { signal })),
+                       /too late/);
+});
+
+test('signal: a clean run does not leave a listener behind', async () => {
+  const ac = new AbortController();
+  await collect(concat([Readable.from(['a'])], { signal: ac.signal }));
+  // Node exposes listener counts on AbortSignal via the events module.
+  const { listenerCount } = await import('node:events');
+  assert.strictEqual(listenerCount(ac.signal, 'abort'), 0);
+});
+
+test('lazy: true forces one-at-a-time over a sync generator', async () => {
+  let created = 0;
+  function* producer() {
+    for (const s of ['a', 'b', 'c']) {
+      created++;
+      yield Readable.from([s]);
+    }
+  }
+
+  for await (const chunk of concat(producer(), { lazy: true })) {
+    assert.strictEqual(String(chunk), 'a');
+    break;
+  }
+  assert.strictEqual(created, 1, `created ${created} sources, expected 1`);
+});
+
+test('lazy: default drains a sync generator, and guards what it finds', async () => {
+  let created = 0;
+  function* producer() {
+    created++;
+    yield slow(['a', 'b', 'c'], 20);
+    created++;
+    yield bombs(15);
+  }
+
+  const it = concat(producer());
+  assert.strictEqual(created, 2); // Drained at call time
+  await assert.rejects(collect(it), /boom/);
+});
+
+test('lazy: false drains a bounded async producer', async () => {
+  async function* producer() {
+    yield Readable.from(['a']);
+    await sleep(5);
+    yield Readable.from(['b']);
+  }
+  assert.deepStrictEqual(await collect(concat(producer(), { lazy: false })), ['a', 'b']);
+});
+
+test('destroyOnReturn: false hands sources back alive', async () => {
+  const b = new Readable({ read() {} });
+
+  for await (const chunk of concat([Readable.from(['a']), b], { destroyOnReturn: false })) {
+    assert.strictEqual(String(chunk), 'a');
+    break;
+  }
+  await sleep(10);
+  assert.ok(!b.destroyed, 'source left alive for the caller');
+  assert.strictEqual(b.listenerCount('error'), 0); // And left unguarded
+  b.destroy();
+});
+
+// ---------------------------------------------------------------------------
+// laziness
+// ---------------------------------------------------------------------------
+
+test('an async producer is pulled one source at a time', async () => {
+  let created = 0;
+  async function* producer() {
+    for (const s of ['a', 'b', 'c']) {
+      created++;
+      yield Readable.from([s]);
+    }
+  }
+
+  for await (const chunk of concat(producer())) {
+    assert.strictEqual(String(chunk), 'a');
+    break;
+  }
+  assert.strictEqual(created, 1, `created ${created} sources, expected 1`);
+});
+
+test('factories in a sync collection stay inert until reached', async () => {
+  let opened = 0;
+  const open = (s) => () => { opened++; return Readable.from([s]); };
+
+  const it = concat([open('a'), open('b'), open('c')]);
+  assert.strictEqual(opened, 0); // Nothing opened at call time
+
+  for await (const chunk of it) {
+    assert.strictEqual(String(chunk), 'a');
+    break;
+  }
+  assert.strictEqual(opened, 1, `opened ${opened}, expected 1`);
+});
+
+test('factories are still guarded once materialized', async () => {
+  await assert.rejects(
+    collect(concat([() => slow(['a', 'b', 'c'], 20), () => bombs(5)])), /boom/);
+});
+
+test('async factories are awaited', async () => {
+  const out = await collect(concat([
+    async () => Readable.from(['a']),
+    async () => { await sleep(5); return Readable.from(['b']); },
+  ]));
+  assert.deepStrictEqual(out, ['a', 'b']);
+});
+
+// ---------------------------------------------------------------------------
+// source variety (ronag's "this doesn't support old streams", jasnell's
+// "make it work with any stream.Readable, ReadableStream, and async iterable")
+// ---------------------------------------------------------------------------
+
+test('supports legacy streams1 sources', async () => {
+  const legacy = new EventEmitter();
+  legacy.readable = true;
+  legacy.pause = () => {};
+  legacy.resume = () => {};
+  legacy.pipe = () => {};
+  setTimeout(() => {
+    legacy.emit('data', 'old');
+    legacy.emit('end');
+  }, 5);
+
+  const out = await collect(concat([legacy, Readable.from(['new'])]));
+  assert.deepStrictEqual(out, ['old', 'new']);
+});
+
+test('supports web ReadableStream sources', async () => {
+  const web = new ReadableStream({
+    start(c) { c.enqueue('w1'); c.enqueue('w2'); c.close(); },
+  });
+  const out = await collect(concat([web, Readable.from(['n'])]));
+  assert.deepStrictEqual(out, ['w1', 'w2', 'n']);
+});
+
+test('a failing web stream surfaces as a rejection', async () => {
+  const web = new ReadableStream({
+    start(c) { c.enqueue('w1'); c.error(new Error('web boom')); },
+  });
+  await assert.rejects(collect(concat([web])), /web boom/);
+});
+
+// ---------------------------------------------------------------------------
+// composition — the point of returning an iterator rather than a stream
+// ---------------------------------------------------------------------------
+
+test('Readable.from(concat(...)) round-trips', async () => {
+  const r = Readable.from(concat([Readable.from(['a']), Readable.from(['b'])]));
+  assert.ok(r instanceof Readable);
+  assert.deepStrictEqual(await collect(r), ['a', 'b']);
+});
+
+test('consumers.text(concat(...)) works', async () => {
+  const value = await text(Readable.from(concat([
+    Readable.from(['hello, ']),
+    Readable.from(['world']),
+  ])));
+  assert.strictEqual(value, 'hello, world');
+});
+
+test('objects pass through untouched — no objectMode to guess', async () => {
+  const out = [];
+  for await (const obj of concat([
+    Readable.from([{ id: 1 }]),
+    Readable.from([{ id: 2 }]),
+  ])) out.push(obj);
+
+  assert.deepStrictEqual(out, [{ id: 1 }, { id: 2 }]);
+});