feat: manage independently resumable call frames in the object graph

crates/core/src/eval.rs

@@ -1,10 +1,11 @@
 use crate::ExecutionProgress;
-use crate::frame::{NativeReturn, StackEntry, SuspendedCall};
+use crate::frame::{Coroutine, CoroutineState, NativeReturn, StackEntry, SuspendedCall};
 use crate::{
     CompareOp, EndorCellObject, EndorCodeObject, EndorFrame, EndorFunctionObject,
     EndorStringObject, ExecutionLimits, ExecutionLocation, Instruction, ObjectRef, Opcode, Value,
     Vm, VmError,
 };
+use crate::{CoroutineProgress, ObjectPayload};
 use crate::{NativeContext, NativeResult};
 
 enum PreparedCall {
@@ -237,6 +238,233 @@ impl Vm {
         Ok(true)
     }
 
+    /// Prepare a managed independently resumable call. Native delegation may
+    /// run during preparation; source instructions start only on resume.
+    /// The returned handle owns one external reference, like every allocation.
+    pub fn new_coroutine(
+        &mut self,
+        function: ObjectRef,
+        arguments: &[Value],
+        limits: ExecutionLimits,
+    ) -> Result<ObjectRef, VmError> {
+        self.check_coroutine_entry()?;
+        self.last_traceback.clear();
+        let mut executed = 0;
+        let prepared = self.prepare_call(function, arguments.to_vec(), &mut executed, limits)?;
+        let state = match prepared {
+            PreparedCall::Frame(frame) => {
+                if limits.max_frames == 0 {
+                    self.clear_frame(frame);
+                    return Err(VmError::ExecutionLimit("frames"));
+                }
+                CoroutineState::Frames(std::sync::Arc::new(vec![frame]))
+            }
+            PreparedCall::Value(value) => CoroutineState::Ready(value),
+        };
+        // Allocate traced edges before releasing the original execution roots.
+        let coroutine = Coroutine {
+            state,
+            limits,
+            executed,
+        };
+        let mut roots = Vec::new();
+        coroutine.trace(&mut |reference| roots.push(reference));
+        let result = self.allocate(coroutine);
+        for root in roots {
+            self.release(root)?;
+        }
+        result
+    }
+
+    fn check_coroutine_entry(&self) -> Result<(), VmError> {
+        if self.has_suspended_call() || self.objects.is_read_only() {
+            return Err(VmError::UnsupportedOperation(
+                "coroutine operation inside read-only execution",
+            ));
+        }
+        Ok(())
+    }
+
+    /// Resume one managed call, preserving cumulative hard limits. Other calls
+    /// and heap writes may execute between Pending results. No prior instruction
+    /// is replayed. Completion returns an owned value and consumes the frames.
+    pub fn resume_coroutine(
+        &mut self,
+        reference: ObjectRef,
+        slice: usize,
+    ) -> Result<CoroutineProgress, VmError> {
+        self.retain(reference)?;
+        let result = self.resume_coroutine_inner(reference, slice);
+        self.release(reference)?;
+        result
+    }
+
+    fn resume_coroutine_inner(
+        &mut self,
+        reference: ObjectRef,
+        slice: usize,
+    ) -> Result<CoroutineProgress, VmError> {
+        self.check_coroutine_entry()?;
+        if slice == 0 {
+            return Err(VmError::InvalidArgument("execution slice must be positive"));
+        }
+        let coroutine = self.payload::<Coroutine>(reference, "coroutine")?;
+        let limits = coroutine.limits;
+        let mut executed = coroutine.executed;
+        let state = match &coroutine.state {
+            CoroutineState::Frames(frames) => CoroutineState::Frames(frames.clone()),
+            CoroutineState::Ready(value) => CoroutineState::Ready(value.clone()),
+            _ => return Err(VmError::InvalidArgument("coroutine is running or finished")),
+        };
+        let mut roots = Vec::new();
+        match &state {
+            CoroutineState::Frames(frames) => {
+                for frame in frames.iter() {
+                    frame.trace_roots(&mut |root| roots.push(root));
+                }
+            }
+            CoroutineState::Ready(Value::Object(root)) => roots.push(*root),
+            _ => {}
+        }
+        // Transfer edges to execution roots transactionally, before replacing
+        // the managed payload. Duplicate edges must retain duplicate roots.
+        let mut retained = Vec::new();
+        for root in roots {
+            if let Err(error) = self.retain(root) {
+                for root in retained {
+                    self.release(root)?;
+                }
+                return Err(error);
+            }
+            retained.push(root);
+        }
+        if let Err(error) = self.replace_payload(
+            reference,
+            Coroutine {
+                state: CoroutineState::Running,
+                limits,
+                executed,
+            },
+        ) {
+            for root in retained {
+                self.release(root)?;
+            }
+            return Err(error);
+        }
+        self.last_traceback.clear();
+        let mut frames = match state {
+            CoroutineState::Frames(frames) => {
+                std::sync::Arc::try_unwrap(frames).expect("private frame storage has one owner")
+            }
+            CoroutineState::Ready(value) => {
+                if let Err(error) = self.replace_payload(
+                    reference,
+                    Coroutine {
+                        state: CoroutineState::Finished,
+                        limits,
+                        executed,
+                    },
+                ) {
+                    self.drop_root(value);
+                    return Err(error);
+                }
+                return Ok(CoroutineProgress::Complete {
+                    value,
+                    instructions: executed,
+                });
+            }
+            _ => unreachable!(),
+        };
+        let slice_end = executed.saturating_add(slice);
+        let result = self.eval_frames(&mut frames, limits, &mut executed, slice_end);
+        if let Ok(None) = result {
+            let frames = std::sync::Arc::new(frames);
+            let mut roots = Vec::new();
+            for frame in frames.iter() {
+                frame.trace_roots(&mut |root| roots.push(root));
+            }
+            let published = self.replace_payload(
+                reference,
+                Coroutine {
+                    state: CoroutineState::Frames(frames),
+                    limits,
+                    executed,
+                },
+            );
+            for root in roots {
+                self.release(root)?;
+            }
+            published?;
+            return Ok(CoroutineProgress::Pending {
+                instructions: executed,
+            });
+        }
+        if result.is_err() {
+            self.last_traceback = frames
+                .iter()
+                .map(|frame| ExecutionLocation {
+                    function: frame.name.clone(),
+                    filename: frame.executable.filename().into(),
+                    first_line: frame.executable.first_line(),
+                    instruction_offset: frame.current_offset,
+                })
+                .collect();
+        }
+        while let Some(frame) = frames.pop() {
+            self.clear_frame(frame);
+        }
+        let finished = self.replace_payload(
+            reference,
+            Coroutine {
+                state: CoroutineState::Finished,
+                limits,
+                executed,
+            },
+        );
+        match result {
+            Ok(Some(value)) => {
+                if let Err(error) = finished {
+                    self.drop_root(value);
+                    return Err(error);
+                }
+                Ok(CoroutineProgress::Complete {
+                    value,
+                    instructions: executed,
+                })
+            }
+            Err(error) => {
+                finished?;
+                Err(error)
+            }
+            Ok(None) => unreachable!(),
+        }
+    }
+
+    /// Cancellation releases managed frame edges; dropping an unreachable
+    /// coroutine has the same ownership effect, including cycles through locals.
+    /// Writes performed by previous quanta are never rolled back.
+    pub fn cancel_coroutine(&mut self, reference: ObjectRef) -> Result<bool, VmError> {
+        self.check_coroutine_entry()?;
+        let coroutine = self.payload::<Coroutine>(reference, "coroutine")?;
+        if matches!(coroutine.state, CoroutineState::Running) {
+            return Err(VmError::InvalidArgument("coroutine is running"));
+        }
+        if matches!(coroutine.state, CoroutineState::Finished) {
+            return Ok(false);
+        }
+        let limits = coroutine.limits;
+        let executed = coroutine.executed;
+        self.replace_payload(
+            reference,
+            Coroutine {
+                state: CoroutineState::Finished,
+                limits,
+                executed,
+            },
+        )?;
+        Ok(true)
+    }
+
     /// Outer-to-inner execution locations from the last failed call/execute.
     /// These are copied metadata, not heap references or language exceptions.
     pub fn last_traceback(&self) -> &[ExecutionLocation] {

crates/core/src/frame.rs

@@ -113,3 +113,74 @@ pub(crate) struct SuspendedCall {
     pub limits: ExecutionLimits,
     pub executed: usize,
 }
+
+/// Independently parked mutable execution. Unlike read-only slicing, it does
+/// not keep a VM-wide heap scope active between quanta.
+#[derive(Debug)]
+pub enum CoroutineProgress {
+    Pending { instructions: usize },
+    Complete { value: Value, instructions: usize },
+}
+#[derive(Debug)]
+pub(crate) enum CoroutineState {
+    Frames(Arc<Vec<EndorFrame>>),
+    Ready(Value),
+    Running,
+    Finished,
+}
+#[derive(Debug)]
+pub(crate) struct Coroutine {
+    pub state: CoroutineState,
+    pub limits: ExecutionLimits,
+    pub executed: usize,
+}
+impl EndorFrame {
+    /// Exactly the references held as execution roots while this frame runs.
+    /// code/globals/builtins remain reachable through its owning function.
+    pub(crate) fn trace_roots(&self, visit: &mut dyn FnMut(ObjectRef)) {
+        visit(self.function);
+        if let Some(reference) = self.return_type.and_then(|kind| kind.reference()) {
+            visit(reference);
+        }
+        for check in &self.native_returns {
+            for reference in check.expected.references() {
+                visit(reference);
+            }
+        }
+        for value in self
+            .localsplus
+            .iter()
+            .flatten()
+            .chain(self.stack.iter().filter_map(|entry| match entry {
+                StackEntry::Value(value) => Some(value),
+                StackEntry::Null => None,
+            }))
+        {
+            if let Value::Object(reference) = value {
+                visit(*reference);
+            }
+        }
+    }
+}
+impl crate::ObjectPayload for Coroutine {
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+    fn kind_name(&self) -> &'static str {
+        "coroutine"
+    }
+    fn reference_assignment_only(&self) -> bool {
+        true
+    }
+    fn trace(&self, visit: &mut dyn FnMut(ObjectRef)) {
+        match &self.state {
+            CoroutineState::Frames(frames) => {
+                for frame in frames.iter() {
+                    frame.trace_roots(visit);
+                }
+            }
+            CoroutineState::Ready(Value::Object(reference)) => visit(*reference),
+            _ => {}
+        }
+    }
+}

crates/core/src/lib.rs

@@ -42,7 +42,9 @@ mod eval;
 mod frame;
 mod vm;
 
-pub use frame::{EndorFrame, ExecutionLimits, ExecutionLocation, ExecutionProgress};
+pub use frame::{
+    CoroutineProgress, EndorFrame, ExecutionLimits, ExecutionLocation, ExecutionProgress,
+};
 // Preserve existing embedding imports; objects owns these definitions.
 pub use objects::ObjectStore;
 pub use objects::{