fix(vm): Preserve sync helper frame when it suspends inside async coro.

Author: dylan-sutton-chavez

compiler/src/main/exports.rs

@@ -231,8 +231,10 @@ pub unsafe extern "C" fn run_push_event(ptr: *const u8, len: u32) -> i32 {
             .find(|(_, h)| matches!(h.state, crate::modules::vm::types::CoroState::WaitingEvent))
             .map(|(i, h)| (i, h.coro));
         if let Some((idx, coro)) = waiter {
-            if let crate::modules::vm::types::HeapObj::Coroutine(_, _, saved_stack, _, _) = vm.heap.get_mut(coro) {
-                if let Some(top) = saved_stack.last_mut() { *top = val; } else { saved_stack.push(val); }
+            if let crate::modules::vm::types::HeapObj::Coroutine(_, _, saved_stack, _, _, sub_frames) = vm.heap.get_mut(coro) {
+                // When a sync helper called from this coro hit `receive()`, the None placeholder lives in the innermost suspended frame's stack — not the outer's. Pick that frame's buffer when present.
+                let target_stack = if let Some(frame) = sub_frames.last_mut() { &mut frame.stack_delta } else { saved_stack };
+                if let Some(top) = target_stack.last_mut() { *top = val; } else { target_stack.push(val); }
             }
             vm.scheduler[idx].state = crate::modules::vm::types::CoroState::Ready;
         } else {
@@ -257,8 +259,9 @@ pub unsafe extern "C" fn set_host_result(handle: u32) -> i32 {
             .find(|(_, h)| matches!(h.state, crate::modules::vm::types::CoroState::WaitingHostCall))
             .map(|(i, h)| (i, h.coro));
         let Some((idx, coro)) = waiter else { return 2; };
-        if let crate::modules::vm::types::HeapObj::Coroutine(_, _, saved_stack, _, _) = vm.heap.get_mut(coro) {
-            if let Some(top) = saved_stack.last_mut() { *top = val; } else { saved_stack.push(val); }
+        if let crate::modules::vm::types::HeapObj::Coroutine(_, _, saved_stack, _, _, sub_frames) = vm.heap.get_mut(coro) {
+            let target_stack = if let Some(frame) = sub_frames.last_mut() { &mut frame.stack_delta } else { saved_stack };
+            if let Some(top) = target_stack.last_mut() { *top = val; } else { target_stack.push(val); }
         }
         vm.scheduler[idx].state = crate::modules::vm::types::CoroState::Ready;
         0

compiler/src/modules/vm/builtins/async_ops.rs

@@ -5,41 +5,88 @@ use super::super::types::*;
 
 impl<'a> VM<'a> {
 
-    /* Resume a coroutine. On yield: persist ip/slots/stack/iters back into it. On return: restore caller stack/iter state. */
+    // Resume coroutine: persist state on yield, restore caller on return. Suspended sync sub-frames run innermost-first, each pushing its result onto the next frame's stack at the Call site.
     pub fn resume_coroutine(&mut self, callee: Val) -> Result<Val, VmErr> {
-        if let HeapObj::Coroutine(ip, saved_slots, saved_stack, fi, saved_iters) = self.heap.get(callee) {
-            let (ip, fi) = (*ip, *fi);
-            let mut fn_slots = saved_slots.clone();
-            let saved_stack_len = self.stack.len();
-            let saved_iter_len = self.iter_stack.len();
-            self.stack.extend_from_slice(&saved_stack.clone());
-            self.iter_stack.extend(saved_iters.clone());
-            let saved_yielded = self.yielded;
-            self.yielded = false;
-            self.depth += 1;
-            let (_, body, _, _) = self.functions[fi];
-            let result = self.exec_from(body, &mut fn_slots, ip);
-            self.depth -= 1;
-            let result = result?;
-            if self.yielded {
-                let resume_ip = self.resume_ip;
-                let remaining = self.stack.split_off(saved_stack_len);
-                let coro_iters: Vec<IterFrame> = self.iter_stack.drain(saved_iter_len..).collect();
-                if let HeapObj::Coroutine(sip, ss, sst, _, si) = self.heap.get_mut(callee) {
-                    *sip = resume_ip;
-                    *ss = fn_slots;
-                    *sst = remaining;
-                    *si = coro_iters;
+        let (outer_ip, mut outer_slots, outer_stack, outer_fi, outer_iters, mut sync_frames) =
+            if let HeapObj::Coroutine(ip, slots, stack, fi, iters, sf) = self.heap.get(callee) {
+                (*ip, slots.clone(), stack.clone(), *fi, iters.clone(), sf.clone())
+            } else {
+                return Err(cold_type("not a coroutine"));
+            };
+
+        let saved_stack_len = self.stack.len();
+        let saved_iter_len = self.iter_stack.len();
+        self.stack.extend_from_slice(&outer_stack);
+        self.iter_stack.extend(outer_iters);
+        let saved_yielded = self.yielded;
+        self.yielded = false;
+        self.depth += 1;
+
+        // Walk frames inside-out, then the outer. `outer_ran` tracks whether `outer_ip` should be overwritten by `resume_ip` on save — a re-yield inside a sync frame leaves the outer pristine.
+        let mut outer_ran = false;
+        let result: Result<Val, VmErr> = 'drive: loop {
+            if let Some(frame) = sync_frames.pop() {
+                let SyncFrame { ip, fi, mut slots, stack_delta, iter_delta } = frame;
+                let frame_stack_base = self.stack.len();
+                let frame_iter_base = self.iter_stack.len();
+                self.stack.extend(stack_delta);
+                self.iter_stack.extend(iter_delta);
+                let (_, body, _, _) = self.functions[fi];
+                match self.exec_from(body, &mut slots, ip) {
+                    Err(e) => break 'drive Err(e),
+                    Ok(val) if self.yielded => {
+                        let new_stack = if self.stack.len() > frame_stack_base { self.stack.split_off(frame_stack_base) } else { Vec::new() };
+                        let new_iter: Vec<IterFrame> = if self.iter_stack.len() > frame_iter_base { self.iter_stack.drain(frame_iter_base..).collect() } else { Vec::new() };
+                        sync_frames.push(SyncFrame {
+                            ip: self.resume_ip, fi, slots,
+                            stack_delta: new_stack, iter_delta: new_iter,
+                        });
+                        // Any deeper sync calls suspended during this exec come back via the VM-level buffer; chain them on top (still innermost-last).
+                        let newer = core::mem::take(&mut self.pending_sync_frames);
+                        sync_frames.extend(newer);
+                        break 'drive Ok(val);
+                    }
+                    Ok(val) => {
+                        // Frame completed; its return value feeds whatever frame (or outer) was waiting at the Call site.
+                        self.push(val);
+                    }
                 }
-                Ok(result)
             } else {
-                self.stack.truncate(saved_stack_len);
-                self.iter_stack.truncate(saved_iter_len);
-                self.yielded = saved_yielded;
-                Ok(result)
+                let (_, body, _, _) = self.functions[outer_fi];
+                outer_ran = true;
+                match self.exec_from(body, &mut outer_slots, outer_ip) {
+                    Err(e) => break 'drive Err(e),
+                    Ok(val) => {
+                        if self.yielded {
+                            let newer = core::mem::take(&mut self.pending_sync_frames);
+                            sync_frames.extend(newer);
+                        }
+                        break 'drive Ok(val);
+                    }
+                }
+            }
+        };
+
+        self.depth -= 1;
+        let result = result?;
+
+        if self.yielded {
+            let resume_ip = if outer_ran { self.resume_ip } else { outer_ip };
+            let remaining = self.stack.split_off(saved_stack_len);
+            let coro_iters: Vec<IterFrame> = self.iter_stack.drain(saved_iter_len..).collect();
+            if let HeapObj::Coroutine(sip, ss, sst, _, si, sf) = self.heap.get_mut(callee) {
+                *sip = resume_ip;
+                *ss = outer_slots;
+                *sst = remaining;
+                *si = coro_iters;
+                *sf = sync_frames;
             }
+            Ok(result)
         } else {
-            Err(cold_type("not a coroutine"))
+            self.stack.truncate(saved_stack_len);
+            self.iter_stack.truncate(saved_iter_len);
+            self.yielded = saved_yielded;
+            Ok(result)
         }
     }
 

compiler/src/modules/vm/dispatch.rs

@@ -169,10 +169,11 @@ impl<'a> VM<'a> {
                 match self.dispatch(chunk, slots, &mut cache, insns, consts, &mut ip) {
                     Ok(None) => {
                         if self.yielded {
-                            // Event yields keep the None placeholder; `run_push_event` overwrites it before resume.
+                            // Event yields keep the None placeholder (overwritten by `run_push_event` before resume). Sync sub-call yields pushed nothing — the helper's return lands on the stack when its frame completes — so don't pop and don't skip the next PopTop.
                             let event_yield = self.pending.event_wait_request;
-                            let val = if event_yield { Val::none() } else { self.pop().unwrap_or(Val::none()) };
-                            self.resume_ip = if !event_yield && ip < n && matches!(insns.get(ip), Some(ins) if ins.opcode == OpCode::PopTop) { ip + 1 } else { ip };
+                            let sub_call_yield = !self.pending_sync_frames.is_empty();
+                            let val = if event_yield || sub_call_yield { Val::none() } else { self.pop().unwrap_or(Val::none()) };
+                            self.resume_ip = if !event_yield && !sub_call_yield && ip < n && matches!(insns.get(ip), Some(ins) if ins.opcode == OpCode::PopTop) { ip + 1 } else { ip };
                             self.live_slots.truncate(slots_base);
                             self.exception_stack.truncate(exc_base);
                             return Ok(val);

compiler/src/modules/vm/gc.rs

@@ -6,6 +6,18 @@ impl<'a> VM<'a> {
     /* Mark all reachable roots then sweep; non-heap Vals are no-op to mark. */
     pub(crate) fn collect(&mut self, current_slots: &[Val]) {
         for &v in &self.stack { self.heap.mark(v); }
+        for sf in &self.pending_sync_frames {
+            for &v in &sf.slots { self.heap.mark(v); }
+            for &v in &sf.stack_delta { self.heap.mark(v); }
+            for fr in &sf.iter_delta {
+                match fr {
+                    IterFrame::Seq { items, .. } => for &v in items { self.heap.mark(v); },
+                    IterFrame::Coroutine(v) => self.heap.mark(*v),
+                    IterFrame::UserDefined(v) => self.heap.mark(*v),
+                    IterFrame::Range { .. } => {}
+                }
+            }
+        }
         for &v in &self.with_stack { self.heap.mark(v); }
         for &v in &self.yields { self.heap.mark(v); }
         for &v in &self.event_queue { self.heap.mark(v); }

compiler/src/modules/vm/handlers/function.rs

@@ -145,12 +145,15 @@ impl<'a> VM<'a> {
         // Generator/coroutine: return a suspended Coroutine instead of running. Both flags are O(1).
         let is_async_fn = self.is_async.get(fi).copied().unwrap_or(false);
         if is_async_fn || body.is_generator {
-            let coro = self.heap.alloc(HeapObj::Coroutine(0, fn_slots, Vec::new(), fi, Vec::new()))?;
+            let coro = self.heap.alloc(HeapObj::Coroutine(0, fn_slots, Vec::new(), fi, Vec::new(), Vec::new()))?;
             self.push(coro);
             self.depth -= 1;
             return Ok(());
         }
 
+        // Snapshot caller-visible depths so we can split the helper's stack/iter contributions out if it suspends mid-body via a yielding builtin.
+        let stack_base = self.stack.len();
+        let iter_base = self.iter_stack.len();
         let yields_before = self.yields.len();
         let (callee_impure, exec_result) = self.run_body_with_frame(fi, body, chunk, &mut fn_slots, slots);
         self.depth -= 1;
@@ -160,6 +163,19 @@ impl<'a> VM<'a> {
         let result = exec_result?;
         if callee_impure { self.mark_impure(); }
 
+        if self.yielded {
+            // Sync helper suspended mid-execution (e.g. `sleep(0)` from inside a sync fn called by an async coro). Stage its frame on the VM-level buffer; `resume_coroutine` drains it onto the enclosing coro so the helper is re-entered from the right ip. Without this, the outer's resume_ip would skip past the unfinished helper and the next StoreName would underflow. A nested sync call inside this helper would already have pushed its own frame first, so the buffer ends up innermost-last.
+            let helper_resume_ip = self.resume_ip;
+            self.resume_ip = 0;
+            let helper_stack_delta = if self.stack.len() > stack_base { self.stack.split_off(stack_base) } else { Vec::new() };
+            let helper_iter_delta: Vec<IterFrame> = if self.iter_stack.len() > iter_base { self.iter_stack.drain(iter_base..).collect() } else { Vec::new() };
+            self.pending_sync_frames.push(SyncFrame {
+                ip: helper_resume_ip, fi, slots: fn_slots,
+                stack_delta: helper_stack_delta, iter_delta: helper_iter_delta,
+            });
+            return Ok(());
+        }
+
         if self.yields.len() > yields_before {
             let fn_yields = self.yields.split_off(yields_before);
             let val = self.heap.alloc(HeapObj::List(Rc::new(RefCell::new(fn_yields))))?;

compiler/src/modules/vm/handlers/mod.rs

@@ -9,7 +9,7 @@ mod methods_helpers;
 
 pub(super) use crate::modules::vm::{
     VM, Val, VmErr, HeapObj, DictMap, cache, ops,
-    types::{cold_depth, cold_type, cold_value, cold_runtime, cold_overflow, eq_vals_with_heap, ffloor}
+    types::{IterFrame, SyncFrame, cold_depth, cold_type, cold_value, cold_runtime, cold_overflow, eq_vals_with_heap, ffloor}
 };
 
 pub(super) use crate::modules::parser::{OpCode, SSAChunk, ssa_strip};

compiler/src/modules/vm/mod.rs

@@ -115,6 +115,8 @@ pub struct VM<'a> {
     pub(crate) sandbox_off: bool,
     pub(crate) with_stack: Vec<Val>,
     pub(crate) pending: Pending,
+    /* Sync helpers that suspended during the current resume; drained into the active Coroutine on yield-save. Lives at VM scope (not `Pending`) because it propagates across dispatch frames, not within one. */
+    pub(crate) pending_sync_frames: Vec<types::SyncFrame>,
     pub(crate) yielded: bool,
     pub(crate) resume_ip: usize,
     pub output: Vec<String>,
@@ -160,6 +162,7 @@ impl<'a> VM<'a> {
             max_calls: limits.calls,
             with_stack: Vec::new(),
             pending: Pending::new(),
+            pending_sync_frames: Vec::new(),
             yielded: false,
             resume_ip: 0,
             strict_input: false,

compiler/src/modules/vm/types/coro.rs

@@ -33,6 +33,16 @@ pub struct CoroutineHandle {
     pub state: CoroState,
 }
 
+// Suspended sync helper frame: a plain user fn called from a coroutine hit a yielding builtin mid-execution, so its state is snapshotted and parked on the enclosing Coroutine. Frames stack innermost-last; resume walks inside-out so each return value lands on the next frame's stack at the Call site.
+#[derive(Clone, Debug)]
+pub struct SyncFrame {
+    pub ip: usize,
+    pub fi: usize,
+    pub slots: Vec<Val>,
+    pub stack_delta: Vec<Val>,
+    pub iter_delta: Vec<IterFrame>,
+}
+
 /* Call-site snapshot for traceback rendering; pushed by `exec_call`, popped on return/error. */
 #[derive(Clone, Debug)]
 pub struct CallFrame {

compiler/src/modules/vm/types/mod.rs

@@ -164,7 +164,8 @@ pub enum HeapObj {
     Property(Val, Val),
     // Intermediate produced by `prop.setter`: callable that takes a function and returns a new `Property` with the setter attached.
     PropertySetter(Val),
-    Coroutine(usize, Vec<Val>, Vec<Val>, usize, Vec<IterFrame>),
+    // Trailing `Vec<SyncFrame>` stacks suspended sync sub-calls (innermost-last): plain user fns called from this coro that hit a yielding builtin before returning. Resume walks inside-out so each return lands on the next frame's stack at the Call site.
+    Coroutine(usize, Vec<Val>, Vec<Val>, usize, Vec<IterFrame>, Vec<SyncFrame>),
     /* Produced by `import m`; attr access via LoadAttr, calls fuse through CallMethod. */
     Module(String, Vec<(String, Val)>),
     /* A native binding lifted to a first-class callable. */
@@ -308,7 +309,7 @@ pub(crate) fn for_each_val(obj: &HeapObj, mut f: impl FnMut(Val)) {
             f(*cls);
             for (k, v) in attrs.borrow().iter() { f(k); f(v); }
         }
-        HeapObj::Coroutine(_, slots, stack, _, iters) => {
+        HeapObj::Coroutine(_, slots, stack, _, iters, sub_frames) => {
             for &v in slots { f(v); }
             for &v in stack { f(v); }
             for fr in iters { match fr {
@@ -317,6 +318,16 @@ pub(crate) fn for_each_val(obj: &HeapObj, mut f: impl FnMut(Val)) {
                 IterFrame::UserDefined(v) => f(*v),
                 IterFrame::Range { .. } => {}
             }}
+            for sf in sub_frames {
+                for &v in &sf.slots { f(v); }
+                for &v in &sf.stack_delta { f(v); }
+                for fr in &sf.iter_delta { match fr {
+                    IterFrame::Seq { items, .. } => for &v in items { f(v); },
+                    IterFrame::Coroutine(v) => f(*v),
+                    IterFrame::UserDefined(v) => f(*v),
+                    IterFrame::Range { .. } => {}
+                }}
+            }
         }
         HeapObj::Func(_, defaults, captures) => {
             for &v in defaults { f(v); }