fix(runtime): V8/Node microtask-tick parity for promise resolution and Web Streams

crates/perry-codegen/src/runtime_decls/strings_part2.rs

@@ -884,6 +884,7 @@ pub(crate) fn declare_phase_b_strings_part2(module: &mut LlModule) {
     module.declare_function("js_promise_resolve", VOID, &[I64, DOUBLE]);
     module.declare_function("js_promise_reject", VOID, &[I64, DOUBLE]);
     module.declare_function("js_promise_resolved", I64, &[DOUBLE]);
+    module.declare_function("js_async_fn_result", I64, &[DOUBLE]);
     module.declare_function("js_promise_rejected", I64, &[DOUBLE]);
     // Issue #100: build a module-namespace object from parallel key/
     // value arrays. Called from `__perry_init_<prefix>` (populate the

crates/perry-codegen/src/stmt/mod.rs

@@ -260,13 +260,14 @@ pub(crate) fn lower_stmt(ctx: &mut FnCtx<'_>, stmt: &Stmt) -> Result<()> {
             }
             let v = lower_return_expr(ctx, e)?;
             // Phase E: async functions wrap their return value in
-            // js_promise_resolved so callers can await the result.
-            // If the value is already a promise (e.g. `return
-            // Promise.resolve(x)`), js_promise_resolved is a no-op
-            // wrap that the caller's await loop unwraps anyway.
+            // js_async_fn_result so callers can await the result. Unlike
+            // js_promise_resolved (whose Promise.resolve(p) === p identity
+            // is spec for Promise.resolve only), an async fn returning a
+            // promise must produce a FRESH promise that adopts the inner
+            // via the two-tick thenable job (V8 microtask-hop parity).
             let final_v = if ctx.is_async_fn {
                 let blk = ctx.block();
-                let handle = blk.call(crate::types::I64, "js_promise_resolved", &[(DOUBLE, &v)]);
+                let handle = blk.call(crate::types::I64, "js_async_fn_result", &[(DOUBLE, &v)]);
                 crate::expr::nanbox_pointer_inline_pub(blk, &handle)
             } else {
                 v

crates/perry-runtime/src/promise/assimilate.rs

@@ -169,10 +169,12 @@ pub(crate) fn promise_resolve_assimilating(promise: *mut Promise, value: f64) {
                 js_promise_resolve(promise, value);
                 return;
             }
-            // No own `then` → intrinsic `then`; native promise→promise wiring.
+            // No own `then` → intrinsic `then`. Adopt via the native job so
+            // the outer settles exactly two ticks later (V8 hop parity; see
+            // `enqueue_native_adoption_job`).
             OwnThen::None => {
                 let inner = crate::value::js_nanbox_get_pointer(value) as *mut Promise;
-                js_promise_resolve_with_promise(promise, inner);
+                enqueue_native_adoption_job(promise, inner);
                 return;
             }
         }
@@ -185,6 +187,82 @@ pub(crate) fn promise_resolve_assimilating(promise: *mut Promise, value: f64) {
     }
 }
 
+/// ECMA-262 hop parity for resolving a promise WITH a native promise (no own
+/// `then`): V8 still runs `PromiseResolveThenableJob` (+1 tick) and the
+/// intrinsic `then` it invokes registers a reaction (+1 tick), so adoption of
+/// an already-settled inner is observable exactly two ticks after resolve —
+/// `new Promise(res => res(Promise.resolve(1))).then(X)` fires `X` on tick 3
+/// in Node, and `async fn` returning a promise behaves identically. The old
+/// synchronous `js_promise_resolve_with_promise` wiring settled the outer
+/// ZERO ticks later, which reordered any promise chain racing the adoption
+/// (Next.js cold-start head reorder: the flight client's module-dep chain
+/// lost exactly one such race against the RSC stream read loop).
+pub(super) fn enqueue_native_adoption_job(outer: *mut Promise, inner: *mut Promise) {
+    use crate::closure::{js_closure_alloc, js_closure_set_capture_ptr};
+
+    let callback = js_closure_alloc(native_promise_adoption_job as *const u8, 2);
+    js_closure_set_capture_ptr(callback, 0, outer as i64);
+    js_closure_set_capture_ptr(callback, 1, inner as i64);
+
+    let context = capture_context();
+    let ids = crate::async_hooks::init_resource(
+        "PromiseResolveThenableJob",
+        f64::from_bits(crate::value::TAG_UNDEFINED),
+        false,
+    );
+    TASK_QUEUE.with(|q| {
+        q.borrow_mut().push_back(Task::Microtask {
+            callback,
+            context,
+            async_id: ids.async_id,
+            trigger_async_id: ids.trigger_async_id,
+        });
+    });
+    crate::event_pump::js_notify_main_thread();
+}
+
+/// Job body — the intrinsic-`then` invocation of the adoption job. For a
+/// settled inner the adoption must land as a REACTION (one more tick), never
+/// a synchronous copy, matching V8's reaction-job. A still-pending inner
+/// falls back to the existing chain wiring: its settlement path already
+/// delivers through the microtask runner.
+extern "C" fn native_promise_adoption_job(closure: *const crate::closure::ClosureHeader) -> f64 {
+    use crate::closure::js_closure_get_capture_ptr;
+
+    let outer = js_closure_get_capture_ptr(closure, 0) as *mut Promise;
+    let inner = js_closure_get_capture_ptr(closure, 1) as *mut Promise;
+    if outer.is_null() || inner.is_null() {
+        return f64::from_bits(crate::value::TAG_UNDEFINED);
+    }
+    let settled = unsafe {
+        match (*inner).state {
+            PromiseState::Fulfilled => Some(((*inner).value, false)),
+            PromiseState::Rejected => Some(((*inner).reason, true)),
+            PromiseState::Pending => None,
+        }
+    };
+    match settled {
+        Some((value, is_error)) => {
+            // Null-closure AsyncStep = a pure propagation task: the runner
+            // resolves/rejects `outer` with `value` on the next tick.
+            TASK_QUEUE.with(|q| {
+                q.borrow_mut().push_back(Task::AsyncStep(
+                    std::ptr::null(),
+                    value,
+                    outer,
+                    is_error,
+                    capture_context(),
+                ));
+            });
+            crate::event_pump::js_notify_main_thread();
+        }
+        None => {
+            super::then::js_promise_resolve_with_promise(outer, inner);
+        }
+    }
+    f64::from_bits(crate::value::TAG_UNDEFINED)
+}
+
 #[inline]
 fn thenable_job_take_guard(guard_arr: *mut crate::array::ArrayHeader) -> bool {
     use crate::array::{js_array_get_f64, js_array_set_f64};

crates/perry-runtime/src/promise/async_step.rs

@@ -380,6 +380,27 @@ pub extern "C" fn js_async_step_chain(value: f64, step_closure: ClosurePtr) -> *
 /// function) or step_closure doesn't match (nested async-fn call,
 /// where the outer activation's `next` must NOT be settled here).
 /// Fall back to `js_promise_resolved(value)`.
+/// Codegen entry for the return value of an await-less async fn (the
+/// transform leaves those un-converted; `Stmt::Return` wraps the value
+/// directly). Differs from `js_promise_resolved` in one spec-visible way: an
+/// async fn always returns a FRESH promise, and `return <promise>` ADOPTS the
+/// inner via the two-tick job (V8 hop parity; hops.js
+/// `asyncfn-return-resolved-promise` = t0 t1 t2 X in Node) — the #2823
+/// `Promise.resolve(p) === p` identity does not apply to async returns.
+#[no_mangle]
+pub extern "C" fn js_async_fn_result(value: f64) -> *mut Promise {
+    let value = adapt_foreign_promise_value(value);
+    if js_value_is_promise(value) != 0 {
+        let inner = crate::value::js_nanbox_get_pointer(value) as *mut Promise;
+        if !inner.is_null() {
+            let fresh = js_promise_new();
+            super::assimilate::enqueue_native_adoption_job(fresh, inner);
+            return fresh;
+        }
+    }
+    js_promise_resolved(value)
+}
+
 #[no_mangle]
 pub extern "C" fn js_async_step_done(value: f64, step_closure: ClosurePtr) -> *mut Promise {
     // PR #1004 followup (sibling to js_async_step_chain): adapt a
@@ -409,6 +430,19 @@ pub extern "C" fn js_async_step_done(value: f64, step_closure: ClosurePtr) -> *m
         trap.trap_next
     } else {
         bump(&MT_STEP_DONE_REUSE_MISS);
+        // An async fn always returns a FRESH promise — `js_promise_resolved`'s
+        // #2823 identity short-circuit (Promise.resolve(p) === p) must not
+        // apply to `return <promise>` from an async fn, which instead adopts
+        // the inner promise via the two-tick job (V8 hop parity; hops.js
+        // `asyncfn-return-resolved-promise` = t0 t1 t2 X in Node).
+        if js_value_is_promise(value) != 0 {
+            let inner = crate::value::js_nanbox_get_pointer(value) as *mut Promise;
+            if !inner.is_null() {
+                let fresh = js_promise_new();
+                super::assimilate::enqueue_native_adoption_job(fresh, inner);
+                return fresh;
+            }
+        }
         js_promise_resolved(value)
     }
 }
@@ -426,11 +460,13 @@ fn resolve_trap_next_with_adoption(target: *mut Promise, value: f64) {
         js_promise_resolve(target, value);
         return;
     }
-    // Native Promise: chain `target` to follow its eventual state.
+    // Native Promise: adopt via the native job — `return <promise>` from an
+    // async fn is observable two ticks later in V8 (hop parity; see
+    // `enqueue_native_adoption_job`).
     if js_value_is_promise(value) != 0 {
         let inner = crate::value::js_nanbox_get_pointer(value) as *mut Promise;
         if !inner.is_null() && inner != target {
-            js_promise_resolve_with_promise(target, inner);
+            super::assimilate::enqueue_native_adoption_job(target, inner);
             return;
         }
     }
@@ -440,7 +476,7 @@ fn resolve_trap_next_with_adoption(target: *mut Promise, value: f64) {
     if assim.to_bits() != value.to_bits() && js_value_is_promise(assim) != 0 {
         let inner = crate::value::js_nanbox_get_pointer(assim) as *mut Promise;
         if !inner.is_null() && inner != target {
-            js_promise_resolve_with_promise(target, inner);
+            super::assimilate::enqueue_native_adoption_job(target, inner);
             return;
         }
     }

crates/perry-runtime/src/promise/then.rs

@@ -1662,7 +1662,22 @@ extern "C" fn finally_passthrough_fulfill(
     let next = js_closure_get_capture_ptr(closure, 0) as *mut Promise;
     let value = js_closure_get_capture_f64(closure, 1);
     if !next.is_null() {
-        js_promise_resolve(next, value);
+        // V8 hop parity: `.finally(cb).then(X)` fires `X` on the FOURTH tick
+        // in Node (hops.js `finally-then`: t0 t1 t2 t3 X) — the spec's
+        // ThenFinally resolves `next` through `promiseResolve(C, cb()).then(
+        // () => value)`, whose then-return propagation costs one more tick
+        // than this passthrough's old direct `js_promise_resolve(next, v)`.
+        // Settle `next` via a propagation task instead.
+        TASK_QUEUE.with(|q| {
+            q.borrow_mut().push_back(Task::AsyncStep(
+                std::ptr::null(),
+                value,
+                next,
+                false,
+                capture_context(),
+            ));
+        });
+        crate::event_pump::js_notify_main_thread();
     }
     f64::from_bits(crate::value::TAG_UNDEFINED)
 }
@@ -1677,7 +1692,17 @@ extern "C" fn finally_passthrough_reject(
     let next = js_closure_get_capture_ptr(closure, 0) as *mut Promise;
     let reason = js_closure_get_capture_f64(closure, 1);
     if !next.is_null() {
-        js_promise_reject(next, reason);
+        // Same extra tick as the fulfilled passthrough (V8 hop parity).
+        TASK_QUEUE.with(|q| {
+            q.borrow_mut().push_back(Task::AsyncStep(
+                std::ptr::null(),
+                reason,
+                next,
+                true,
+                capture_context(),
+            ));
+        });
+        crate::event_pump::js_notify_main_thread();
     }
     f64::from_bits(crate::value::TAG_UNDEFINED)
 }

crates/perry-stdlib/src/streams.rs

@@ -58,7 +58,7 @@ mod transform;
 mod writable;
 
 pub use tee::js_readable_stream_tee;
-use tee::{tee_branches_of, tee_close_branches, tee_deliver, tee_error_branches, tee_source_of};
+use tee::{tee_branches_of, tee_error_branches, tee_source_of};
 
 pub use self::byob::{
     js_readable_stream_controller_byob_request, js_readable_stream_get_byob_reader,
@@ -704,7 +704,9 @@ pub(super) unsafe fn maybe_pull(stream_id: usize) {
     // `pull_cb` of its own; the source's enqueue fans the chunk out to both
     // branches. `force` so a highWaterMark-0 flight producer actually pulls.
     if let Some(source) = tee_source_of(stream_id) {
-        maybe_pull_force(source);
+        // Tick parity: the pull travels a microtask cycle (buffered chunks +
+        // close discovery included); a live producer is driven from that job.
+        tee::tee_schedule_pull(source);
         return;
     }
     maybe_pull_inner(stream_id, false);
@@ -1512,11 +1514,10 @@ pub unsafe extern "C" fn js_readable_stream_controller_enqueue(
             "The \"buffer\" argument must be an instance of Buffer, TypedArray, or DataView",
         );
     }
-    // #5989: a tee'd source never queues its own chunks — each enqueue fans out
-    // to BOTH branches (delivering to a parked read or queueing per branch).
-    if let Some((a, b)) = tee_branches_of(id) {
-        tee_deliver(a, chunk_bits, is_byte_stream);
-        tee_deliver(b, chunk_bits, is_byte_stream);
+    // #5989 + Node tick parity: a tee'd source's enqueue stays in the SOURCE
+    // queue; branches receive chunks only through the demand-driven pull
+    // cycle (see `tee::tee_source_enqueue`).
+    if tee::tee_source_enqueue(id, chunk, chunk_bits, is_byte_stream) {
         return f64::from_bits(TAG_UNDEFINED);
     }
     // A pending BYOB read (byte streams only) takes the chunk before the
@@ -1577,8 +1578,10 @@ pub unsafe extern "C" fn js_readable_stream_controller_close(stream_handle: f64)
             eprintln!("[STREAM] controller_close stream={id:#x}");
         }
     }
-    // #5989: closing a tee'd source closes both branches (each drains its queue
-    // first) rather than the source's own — absent — reader.
+    // #5989 + Node tick parity: closing a tee'd source marks it Closed; the
+    // branch close is DISCOVERED by the pull cycle draining the source queue
+    // empty (byte tee: prompt; default tee: nextTick-deferred) — never
+    // delivered eagerly ahead of undrained chunks.
     if tee_branches_of(id).is_some() {
         {
             let mut g = READABLE_STREAMS.lock().unwrap();
@@ -1588,7 +1591,7 @@ pub unsafe extern "C" fn js_readable_stream_controller_close(stream_handle: f64)
                 }
             }
         }
-        tee_close_branches(id);
+        tee::tee_schedule_pull(id);
         return f64::from_bits(TAG_UNDEFINED);
     }
     {
@@ -1774,6 +1777,10 @@ pub unsafe extern "C" fn js_reader_read(reader_handle: f64) -> *mut Promise {
             None => Some((TAG_UNDEFINED, true, false)),
         }
     };
+    // Consumer progress: if this readable is a transform's output and its
+    // queue just drained (pop emptied it, or the read parked on an empty
+    // queue), release write promises parked on backpressure.
+    transform::transform_release_writes(stream_id);
     if let Some((reader_id, reason)) = closed_rejection {
         let p = READERS
             .lock()

crates/perry-stdlib/src/streams/byob.rs

@@ -100,6 +100,16 @@ unsafe fn view_info(view_bits: u64) -> Option<ViewInfo> {
 
 /// `chunk.byteLength` for desiredSize accounting on byte streams; 1.0 for
 /// values whose byte length can't be derived (matches the count fallback).
+/// Clone a byte chunk as a fresh Uint8Array (spec `CloneAsUint8Array`, used by
+/// the byte-stream tee so branches never share a mutable buffer). Non-byte
+/// values pass through unchanged.
+pub(super) unsafe fn clone_byte_chunk(chunk_bits: u64) -> u64 {
+    match read_bytes_from_chunk(chunk_bits) {
+        Some(bytes) => alloc_uint8array_from_bytes(&bytes),
+        None => chunk_bits,
+    }
+}
+
 pub(super) unsafe fn chunk_byte_length(chunk_bits: u64) -> f64 {
     match read_bytes_from_chunk(chunk_bits) {
         Some(bytes) => bytes.len() as f64,
@@ -302,6 +312,10 @@ pub unsafe extern "C" fn js_reader_read_with_view(reader_handle: f64, view: f64)
     }
 
     let filled = fill_view_from_queue(stream_id, &info);
+    // A BYOB drain (or an about-to-park read on an empty queue) is consumer
+    // progress on this readable — release transform writes parked on
+    // backpressure, mirroring the default-reader path in `js_reader_read`.
+    super::transform::transform_release_writes(stream_id);
     if filled > 0 {
         let bytes = std::slice::from_raw_parts(info.data, filled);
         let value = alloc_view_of_kind(info.kind, info.elem_size, bytes);