stream: defer readable async iterator listener

Defer installing the readable listener for async iteration until read()
returns null. This avoids listener dispatch on the common path where
data is already buffered while preserving end-of-stream error handling.

Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5

Author: trivikr

lib/internal/streams/readable.js

@@ -1380,6 +1380,7 @@ function streamToAsyncIterator(stream, options) {
 
 async function* createAsyncIterator(stream, options) {
   let callback = nop;
+  let isListening = false;
 
   function next(resolve) {
     if (this === stream) {
@@ -1390,15 +1391,18 @@ async function* createAsyncIterator(stream, options) {
     }
   }
 
-  stream.on('readable', next);
-
   let error;
   const cleanup = eos(stream, { writable: false }, (err) => {
     error = err ? aggregateTwoErrors(error, err) : null;
     callback();
     callback = nop;
   });
 
+  function setupReadable() {
+    isListening = true;
+    stream.on('readable', next);
+  }
+
   try {
     while (true) {
       const chunk = stream.destroyed ? null : stream.read();
@@ -1409,6 +1413,9 @@ async function* createAsyncIterator(stream, options) {
       } else if (error === null) {
         return;
       } else {
+        if (!isListening) {
+          setupReadable();
+        }
         await new Promise(next);
       }
     }