stream/iter: fused stateless transforms do not normalize output between transforms

[trivikr]

Version

main

Platform

N/A

Subsystem

stream

What steps will reproduce the bug?

import { from, fromSync, pull, pullSync, text, textSync } from 'node:stream/iter';

const first = (chunks) =>
  chunks === null ? null : new Uint8Array([65]);

const asyncSecond = (chunks) => {
  if (chunks !== null)
    console.log('async received array:', Array.isArray(chunks));
  return chunks;
};

const syncSecond = (chunks) => {
  if (chunks !== null)
    console.log('sync received array:', Array.isArray(chunks));
  return chunks;
};

await text(pull(from('x'), first, asyncSecond));
textSync(pullSync(fromSync('x'), first, syncSecond));

How often does it reproduce? Is there a required condition?

Always

What is the expected behavior? Why is that the expected behavior?

async received array: true
sync received array: true

The first transform’s Uint8Array result should be normalized to a single-element Uint8Array[] batch before it is passed to the second transform.

What do you see instead?

async received array: false
sync received array: false

The second transform receives the raw Uint8Array returned by the first transform. Both fused async and synchronous paths normalize only after every transform in the run has executed.

Additional information

No response