//! VM 的指令执行与调用生命周期；宿主入口在此准备帧，底层对象规则仍通过 Vm/Store 操作。
//! 执行结果中的对象拥有一份外部根，参数通常借用；帧/原生上下文负责保留和清理执行期间根。
//! 只读续执行跨 Pending 保持 Heap 作用域；托管可变协程另存载荷，不沿用该全局作用域。

use crate::ExecutionProgress;
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};

/// prepare_call 统一结果，成功的 Value/Frame 都携带执行期间已取得的拥有根。
enum PreparedCall {
    // 原生或委托调用在准备阶段直接完成，结果根移交上层。
    Value(Value),
    // 源码帧已绑定参数并持有根，交给循环执行或暂存。
    Frame(EndorFrame),
}

/// 求值器构造携带代码单元偏移的字节码错误，不将偏移误报为源码行。
fn invalid(offset: usize, reason: &'static str) -> VmError {
    VmError::InvalidBytecode { offset, reason }
}

impl Vm {
    /// Execute a code object in a globals dictionary. Like CPython's EvalCode
    /// entry path, this constructs a temporary function and enters its frame.
    /// Object results own one external reference: release_value() when done.
    /// 宿主执行已验证代码，使用默认预算，最终对象结果归调用方拥有。
    pub fn execute(&mut self, code: ObjectRef, globals: ObjectRef) -> Result<Value, VmError> {
        self.execute_with_limits(code, globals, ExecutionLimits::default())
    }

    /// 宿主以指定预算执行代码，临时源码函数包装全局/代码引用，调用结束后释放包装根。
    pub fn execute_with_limits(
        &mut self,
        code: ObjectRef,
        globals: ObjectRef,
        limits: ExecutionLimits,
    ) -> Result<Value, VmError> {
        // 新调用从空诊断开始，后续失败才写入该次剩余帧回溯。
        self.last_traceback.clear();
        // 借用代码载荷的诊断名后转为拥有 String，避免跨可变分配保留载荷借用。
        let name = self
            .payload::<EndorCodeObject>(code, "code")?
            .code()
            .name()
            .to_owned();
        // 创建无参数默认值/无捕获的临时执行函数；代码所需布局仍由函数校验。
        let function = self.new_function(name, code, globals, Vec::new(), Vec::new())?;
        // 先保存调用结果，使成功和错误都能在返回前释放临时函数根。
        let result = self.call_with_limits(function, &[], limits);
        // 结果对象已由调用路径保留，释放包装函数不应使返回结果失效。
        self.release(function)?;
        result
    }

    /// Dispatch both language and host functions. Arguments are borrowed;
    /// every Object result owns one external reference, including native
    /// callbacks that return one of their borrowed arguments.
    /// 宿主同步调用源码/原生函数，参数借用，返回对象根需 release_value。
    pub fn call(&mut self, function: ObjectRef, arguments: &[Value]) -> Result<Value, VmError> {
        self.call_with_limits(function, arguments, ExecutionLimits::default())
    }

    /// Allow result allocation while rejecting mutations of objects that were
    /// already alive on entry. All reads share the call's normal frame budget.
    /// Returned read handles are borrowed dependency identities, not owned roots.
    /// 宿主同步只读兼容入口，只保留对象读取集合，丢弃精确依赖键。
    pub fn call_read_only_with_limits(
        &mut self,
        function: ObjectRef,
        arguments: &[Value],
        limits: ExecutionLimits,
    ) -> Result<(Value, Vec<ObjectRef>), VmError> {
        // 精确入口统一维护只读作用域与结果所有权，这里只改变返回观察的形状。
        let (value, reads) =
            self.call_read_only_precise_with_limits(function, arguments, limits)?;
        Ok((value, reads.objects))
    }

    /// Synchronous read-only call with precise observations as well as the
    /// conservative object list. Result ownership and hard limits match call().
    /// ReadSet and all identities/value snapshots inside it are borrowed only.
    /// 宿主同步只读入口，建立保护快照，成功返回拥有结果与借用观察集合。
    pub fn call_read_only_precise_with_limits(
        &mut self,
        function: ObjectRef,
        arguments: &[Value],
        limits: ExecutionLimits,
    ) -> Result<(Value, crate::ReadSet), VmError> {
        // 快照进入前对象身份；之后新建结果允许初始化，但已有对象受写屏障保护。
        let scope = self.objects.begin_read_only();
        // 先保存调用结果，使成功和错误都能在返回前释放临时函数根。
        let result = self.call_with_limits(function, arguments, limits);
        // 结束域先取回借用观察，之后才将调用成功值与观察组合返回。
        let reads = self.objects.end_read_only_precise(scope)?;
        result.map(|value| (value, reads))
    }

    /// 同步调用核心：准备可立即完成的原生结果或源码帧，执行后清理帧及保存错误回溯。
    pub fn call_with_limits(
        &mut self,
        function: ObjectRef,
        arguments: &[Value],
        limits: ExecutionLimits,
    ) -> Result<Value, VmError> {
        // 已有只读续执行保留活跃作用域，禁止另开同步调用干扰它。
        if self.suspended_call.is_some() {
            return Err(VmError::UnsupportedOperation(
                "call while read-only execution is suspended",
            ));
        }
        // 新调用从空诊断开始，后续失败才写入该次剩余帧回溯。
        self.last_traceback.clear();
        // 累计调用准备和指令工作量，切片恢复不会将其清零。
        let mut executed = 0;
        // 原生调用可能在准备阶段完成并直接移交拥有结果，源码则进入帧循环。
        let frame = match self.prepare_call(function, arguments.to_vec(), &mut executed, limits)? {
            PreparedCall::Value(value) => return Ok(value),
            PreparedCall::Frame(frame) => frame,
        };
        // 源码帧已创建但预算不允许执行，必须先清理所持根再报错。
        if limits.max_frames == 0 {
            self.clear_frame(frame);
            return Err(VmError::ExecutionLimit("frames"));
        }
        // 显式帧向量代替 Rust 递归调用，调用深度由硬预算控制。
        let mut frames = vec![frame];
        // 同步入口没有软切片限制且不允许 Pending await；只读恢复另按 slice_end 执行。
        let result = self
            .eval_frames(&mut frames, limits, &mut executed, usize::MAX, false)
            .map(|value| value.expect("unbounded slice completes or fails"));
        // 错误时先保存诊断，再销毁帧；使用 current_offset 而不是下一指令偏移。
        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();
        }
        // 逐帧回收栈、局部、函数和返回约束根，不把 Rust drop 误当托管 release。
        while let Some(frame) = frames.pop() {
            self.clear_frame(frame);
        }
        result
    }

    /// Begin a resumable read-only call. One VM owns at most one continuation.
    /// A slice is a soft instruction quantum: native callbacks and delegation
    /// complete atomically and remain subject to the cumulative hard limits.
    /// 宿主开始单个可分片只读调用，原生准备可立即完成，源码帧停放在 VM 内。
    pub fn start_read_only_call(
        &mut self,
        function: ObjectRef,
        arguments: &[Value],
        limits: ExecutionLimits,
        slice: usize,
    ) -> Result<ExecutionProgress, VmError> {
        // 拒绝零量子，避免调度者永远不推进且仍报告 Pending。
        if slice == 0 {
            return Err(VmError::InvalidArgument("execution slice must be positive"));
        }
        // 只读续执行只能在没有其他续执行或只读域时启动，避免作用域令牌交错。
        if self.suspended_call.is_some() || self.objects.is_read_only() {
            return Err(VmError::UnsupportedOperation(
                "start suspended call inside another execution",
            ));
        }
        // 新调用从空诊断开始，后续失败才写入该次剩余帧回溯。
        self.last_traceback.clear();
        // 快照进入前对象身份；之后新建结果允许初始化，但已有对象受写屏障保护。
        let scope = self.objects.begin_read_only();
        // 累计调用准备和指令工作量，切片恢复不会将其清零。
        let mut executed = 0;
        // 准备原生调用/委托可能消耗预算或报错，先捕获结果以便三种分支都结束/保存作用域。
        let prepared = self.prepare_call(function, arguments.to_vec(), &mut executed, limits);
        // 成功直接值关闭域；源码帧保存域；准备失败关闭域后传播错误。
        match prepared {
            // 原生立即完成，无源码帧要保存，但仍需结束观察收集。
            Ok(PreparedCall::Value(value)) => {
                // 若读取域收尾失败，必须归还已经取得的结果根，不能泄漏。
                let reads = match self.objects.end_read_only_precise(scope) {
                    Ok(reads) => reads,
                    // 出错分支在返回前处理本分支拥有的结果/作用域，不能保留半完成状态。
                    Err(error) => {
                        self.drop_root(value);
                        return Err(error);
                    }
                };
                Ok(ExecutionProgress::Complete {
                    value,
                    dependencies: reads.dependencies,
                    reads: reads.objects,
                    instructions: executed,
                })
            }
            // 源帧尚未执行，先通过帧预算再交给 VM 的单个续执行槽。
            Ok(PreparedCall::Frame(frame)) => {
                // 源码帧已创建但预算不允许执行，必须先清理所持根再报错。
                if limits.max_frames == 0 {
                    self.clear_frame(frame);
                    self.objects.end_read_only(scope)?;
                    return Err(VmError::ExecutionLimit("frames"));
                }
                // 保存根仍由帧持有，Heap 只读作用域跨后续 Pending 不关闭。
                self.suspended_call = Some(SuspendedCall {
                    frames: vec![frame],
                    scope,
                    limits,
                    executed,
                });
                // 首片边界从零计，调用准备已用计数也占首片软预算，不另赠一片。
                self.resume_read_only_to(slice)
            }
            // 出错分支在返回前处理本分支拥有的结果/作用域，不能保留半完成状态。
            Err(error) => {
                self.objects.end_read_only(scope)?;
                Err(error)
            }
        }
    }

    /// 调度者判断是否有只读续执行，不表示托管协程数量或等待状态。
    pub fn has_suspended_call(&self) -> bool {
        self.suspended_call.is_some()
    }

    /// Resume existing frames, never replay earlier instructions. The read-only
    /// scope remains active across Pending; preexisting heap writes stay barred.
    /// 宿主从已保存指令位置继续，预算跨片累计，不能重放已执行语句。
    pub fn resume_read_only_call(&mut self, slice: usize) -> Result<ExecutionProgress, VmError> {
        // 拒绝零量子，避免调度者永远不推进且仍报告 Pending。
        if slice == 0 {
            return Err(VmError::InvalidArgument("execution slice must be positive"));
        }
        // 读取此前累计计数，新增软边界以 saturating_add 防整数绕回。
        let executed = self
            .suspended_call
            .as_ref()
            .ok_or(VmError::InvalidArgument("no suspended call"))?
            .executed;
        self.resume_read_only_to(executed.saturating_add(slice))
    }

    /// start/resume 共用的累计软边界执行器；Pending 原样放回状态，终态关闭只读域。
    fn resume_read_only_to(&mut self, slice_end: usize) -> Result<ExecutionProgress, VmError> {
        // 暂时取走续执行状态独占推进；只有 Pending 才重新放回，终态不残留槽。
        let mut call = self
            .suspended_call
            .take()
            .ok_or(VmError::InvalidArgument("no suspended call"))?;
        // 按累计 slice_end 推进现有帧；false 禁止 await 停放，None 表示软切片结束。
        let result = self.eval_frames(
            &mut call.frames,
            call.limits,
            &mut call.executed,
            slice_end,
            false,
        );
        // 软切片耗尽：保留原帧/作用域/硬预算计数，不清理根或重新准备函数。
        if let Ok(None) = result {
            // 报告从调用起点累计的工作量，不是当前片局部数量。
            let instructions = call.executed;
            // 将完整续执行状态归还 VM，原有只读屏障继续生效。
            self.suspended_call = Some(call);
            return Ok(ExecutionProgress::Pending { instructions });
        }
        // 错误时先保存诊断，再销毁帧；使用 current_offset 而不是下一指令偏移。
        if result.is_err() {
            self.last_traceback = call
                .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();
        }
        // 逐帧回收栈、局部、函数和返回约束根，不把 Rust drop 误当托管 release。
        while let Some(frame) = call.frames.pop() {
            self.clear_frame(frame);
        }
        // 完成/失败统一关闭作用域，保留收尾错误供下方与执行结果一起处理。
        let reads = self.objects.end_read_only_precise(call.scope);
        // 仅执行与收尾均成功才交付结果；收尾失败释放成功结果，执行失败优先传播执行错误。
        match (result, reads) {
            (Ok(Some(value)), Ok(reads)) => Ok(ExecutionProgress::Complete {
                value,
                dependencies: reads.dependencies,
                reads: reads.objects,
                instructions: call.executed,
            }),
            // 已产生的成功结果此时不能交付，归还其对象根避免泄漏。
            (Ok(Some(value)), Err(error)) => {
                self.drop_root(value);
                Err(error)
            }
            (Err(error), _) => Err(error),
            (Ok(None), _) => unreachable!(),
        }
    }

    /// Release all retained execution roots and close the read-only scope.
    /// 宿主放弃当前只读续执行，逐帧释放根并关闭作用域，无任务时返回 false。
    pub fn cancel_read_only_call(&mut self) -> Result<bool, VmError> {
        // 没有续执行是正常无操作；有任务则一次取走，后续取消不会重复释放。
        let Some(mut call) = self.suspended_call.take() else {
            return Ok(false);
        };
        // 逐帧回收栈、局部、函数和返回约束根，不把 Rust drop 误当托管 release。
        while let Some(frame) = call.frames.pop() {
            self.clear_frame(frame);
        }
        self.objects.end_read_only(call.scope)?;
        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.
    /// 宿主任务系统准备独立可恢复调用，返回一份协程拥有根；原生准备可发生在首次 resume 之前。
    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;
        // prepare_call 可能直接完成原生调用或返回已持根的源码帧。
        let prepared = self.prepare_call(function, arguments.to_vec(), &mut executed, limits)?;
        // 源码帧暂存 Frames，立即结果暂存 Ready；两者尚持有执行根待转换。
        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),
        };
        // 先建立托管载荷强边，再归还准备阶段根，确保转换期间对象始终存活。
        // 把准备状态包装为托管载荷，trace 会为其保留与帧根一致的内部强边。
        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> {
        // 可变协程不持有全局只读域，也不能干扰 VM 已停放的只读调用。
        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
    }

    /// 恢复状态、把载荷边转为执行根，执行后再停放或提交 Finished；外层负责保护协程本身。
    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"));
        }
        // 把准备状态包装为托管载荷，trace 会为其保留与帧根一致的内部强边。
        let coroutine = self.payload::<Coroutine>(reference, "coroutine")?;
        // 硬限制来自创建时状态，不能在每个量子重新给予预算。
        let limits = coroutine.limits;
        // 从上次停放的累计计数继续，而不是从零执行。
        let mut executed = coroutine.executed;
        // 只复制状态表示/Arc，不因 clone 增加托管计数；先由旧载荷边保证存活。
        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();
        // 列出恢复为活跃执行时所需全部根；Ready 标量没有对象根。
        match &state {
            // 帧身份沿同一 trace 清单转换，不复制执行行为或重新绑定实参。
            CoroutineState::Frames(frames) => {
                // 复用 frame::trace_roots，确保与活跃帧清理清单完全对应。
                for frame in frames.iter() {
                    frame.trace_roots(&mut |root| roots.push(root));
                }
            }
            CoroutineState::Ready(Value::Object(root)) => roots.push(*root),
            _ => {}
        }
        // 恢复时方向相反：先取得每份执行根，再替换旧载荷以释放内部边；重复边对应重复根。
        // 事务式取得新执行根；中途失败仅归还本轮已成功 retain 的部分。
        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();
        // Running 替换释放旧载荷 Arc 后，快照 Arc 应成为唯一所有者，可取回帧向量。
        let mut frames = match state {
            // 帧身份沿同一 trace 清单转换，不复制执行行为或重新绑定实参。
            CoroutineState::Frames(frames) => {
                std::sync::Arc::try_unwrap(frames).expect("private frame storage has one owner")
            }
            // 准备阶段已完成，直接提交 Finished，再把新取得的结果根交给调用者。
            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!(),
        };
        // 软边界按累计计数增加，saturating_add 防边界绕回，硬上限仍独立检查。
        let slice_end = executed.saturating_add(slice);
        // 允许 await 停放；None 既可能是软量子结束，也可能等待对象 Pending。
        let result = self.eval_frames(&mut frames, limits, &mut executed, slice_end, true);
        // 需要停放：先分析是否 await，再把完整帧/计数提交回托管载荷。
        if let Ok(None) = result {
            // waiting 只取最内层帧等待对象；栈顶操作数留存，恢复时不会重放创建它的调用。
            let waiting = frames.last().filter(|frame| frame.waiting).map(|frame| {
                // 已执行 Await 证明栈顶是对象，否则属于执行器内部不变量破坏。
                match Self::top_value(frame).expect("await operand remains on stack") {
                    Value::Object(reference) => reference,
                    _ => unreachable!(),
                }
            });
            // 把帧向量重新装进共享快照，内部边接管前仍持原执行根。
            let frames = std::sync::Arc::new(frames);
            // 枚举每一份根身份，重复项表示多份所有权，不能去重。
            let mut roots = Vec::new();
            // 复用 frame::trace_roots，确保与活跃帧清理清单完全对应。
            for frame in frames.iter() {
                frame.trace_roots(&mut |root| roots.push(root));
            }
            // 先提交 Frames 与完整强边，发布成功前不能释放原执行根造成悬空。
            let published = self.replace_payload(
                reference,
                Coroutine {
                    state: CoroutineState::Frames(frames),
                    limits,
                    executed,
                },
            );
            // 逐项归还原执行根；成功路径已由载荷持边，错误路径则释放准备状态资源。
            for root in roots {
                self.release(root)?;
            }
            // 归还原执行根后传播发布错误；失败不会交付虚假的 Pending 状态。
            published?;
            // Waiting 只返回借用等待身份，图由停放协程帧维持，调度者须保留协程根。
            if let Some(awaitable) = waiting {
                return Ok(CoroutineProgress::Waiting {
                    awaitable,
                    instructions: executed,
                });
            }
            return Ok(CoroutineProgress::Pending {
                instructions: executed,
            });
        }
        // 在帧清理前提取回溯，保存 current_offset 而非下一指令位置。
        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);
        }
        // 无论成功或失败都清空运行状态，保留预算/计数供诊断，Finished 不再持帧图。
        let finished = self.replace_payload(
            reference,
            Coroutine {
                state: CoroutineState::Finished,
                limits,
                executed,
            },
        );
        // 成功必须同时完成状态提交；提交失败归还结果，执行错误路径仍须处理收尾结果。
        match result {
            // 结果拥有根仅在 Finished 提交成功后移交给宿主。
            Ok(Some(value)) => {
                // 结果已产生但状态提交失败，不能泄漏该结果根。
                if let Err(error) = finished {
                    self.drop_root(value);
                    return Err(error);
                }
                Ok(CoroutineProgress::Complete {
                    value,
                    instructions: executed,
                })
            }
            // 此处先确保 Finished 发布成功，再返回原执行错误。
            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.
    /// 宿主取消停放帧或准备结果，用 Finished 替换以释放载荷边；不回滚此前写入。
    pub fn cancel_coroutine(&mut self, reference: ObjectRef) -> Result<bool, VmError> {
        // 先拒绝只读上下文，避免用托管协程绕过活跃只读保护。
        self.check_coroutine_entry()?;
        // 把准备状态包装为托管载荷，trace 会为其保留与帧根一致的内部强边。
        let coroutine = self.payload::<Coroutine>(reference, "coroutine")?;
        // 正执行状态拒绝取消，避免在 Rust 回调重入时移除执行者状态。
        if matches!(coroutine.state, CoroutineState::Running) {
            return Err(VmError::InvalidArgument("coroutine is running"));
        }
        // 重复取消是正常 false，不再修改载荷版本。
        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] {
        &self.last_traceback
    }

    pub fn release_value(&mut self, value: Value) -> Result<(), VmError> {
        if let Value::Object(reference) = value {
            self.release(reference)?;
        }
        Ok(())
    }

    fn retain_value(&mut self, value: &Value) -> Result<(), VmError> {
        self.validate_value(value)?;
        if let Value::Object(reference) = value {
            self.retain(*reference)?;
        }
        Ok(())
    }

    fn drop_root(&mut self, value: Value) {
        self.release_value(value)
            .expect("execution owns every object root it releases");
    }

    fn clear_frame(&mut self, frame: EndorFrame) {
        self.clear_native_returns(frame.native_returns);
        for entry in frame.stack {
            if let StackEntry::Value(value) = entry {
                self.drop_root(value);
            }
        }
        for value in frame.localsplus.into_iter().flatten() {
            self.drop_root(value);
        }
        self.release(frame.function)
            .expect("frame owns its function root");
        if let Some(reference) = frame.return_type.and_then(|kind| kind.reference()) {
            self.release(reference)
                .expect("frame owns its return type root");
        }
    }

    fn prepare_frame(
        &mut self,
        function: ObjectRef,
        arguments: &[Value],
    ) -> Result<EndorFrame, VmError> {
        let data = self
            .payload::<EndorFunctionObject>(function, "function")?
            .clone();
        let executable = self
            .payload::<EndorCodeObject>(data.code(), "code")?
            .shared_code();
        let maximum = executable.parameters().len();
        let required = maximum - data.defaults().len();
        if arguments.len() < required || arguments.len() > maximum {
            return Err(VmError::ArgumentCount {
                name: data.name().into(),
                required,
                maximum,
                provided: arguments.len(),
            });
        }
        for value in arguments {
            self.validate_value(value)?;
        }
        let receiver = arguments
            .first()
            .or_else(|| {
                if required == 0 {
                    data.defaults().first()
                } else {
                    None
                }
            })
            .and_then(|value| match value {
                Value::Object(receiver) => Some(*receiver),
                _ => None,
            });
        let typed_method = if let Some(receiver) = receiver {
            self.objects
                .check_method_contract_call(function, receiver, self.source_type_checks)?
        } else {
            false
        };
        let mut return_type = None;
        if (self.source_type_checks || typed_method)
            && let Some(signature) = executable.source_signature()
        {
            for (index, expected_name) in signature.parameters.iter().enumerate() {
                let value = arguments
                    .get(index)
                    .unwrap_or_else(|| &data.defaults()[index - required]);
                let expected = self
                    .objects
                    .annotation_type(function, expected_name, receiver)?;
                if !self.objects.matches_annotation(value, expected)? {
                    return Err(VmError::TypeMismatch {
                        function: data.name().into(),
                        position: format!("parameter {}", executable.parameters()[index]),
                        expected: expected_name.clone(),
                        actual: self.objects.value_type_name(value)?,
                    });
                }
            }
            return_type = Some(self.objects.annotation_type(
                function,
                &signature.return_type,
                receiver,
            )?);
        }
        self.retain(function)?;
        if let Some(reference) = return_type.and_then(|kind| kind.reference())
            && let Err(error) = self.retain(reference)
        {
            self.release(function)?;
            return Err(error);
        }
        let mut frame = EndorFrame {
            return_type,
            native_returns: Vec::new(),
            function,
            previous: None,
            code: data.code(),
            globals: data.globals(),
            builtins: data.builtins(),
            name: data.name().into(),
            instruction_offset: 0,
            current_offset: 0,
            waiting: false,
            localsplus: vec![None; executable.slot_count()],
            stack: Vec::with_capacity(executable.stack_size()),
            executable,
        };
        let initialized = (|| -> Result<(), VmError> {
            if let Some(receiver) = receiver {
                for (name, expected) in self.objects.method_trait_returns(function, receiver)? {
                    self.push_native_return(
                        &mut frame.native_returns,
                        NativeReturn { name, expected },
                    )?;
                }
            }
            let defaults = data.defaults().iter().skip(arguments.len() - required);
            for (slot, value) in frame
                .localsplus
                .iter_mut()
                .zip(arguments.iter().chain(defaults))
            {
                self.retain_value(value)?;
                *slot = Some(value.clone());
            }
            // Captured local cells occupy the original local slots, as in
            // CPython's localsplus layout. Free cells follow the local slots.
            let local_count = frame.executable.layout().locals.len();
            for (index, slot) in frame.localsplus[..local_count].iter_mut().enumerate() {
                if frame.executable.is_cell_slot(index) {
                    let cell = self.new_cell(slot.clone())?;
                    if let Some(old) = slot.replace(Value::Object(cell)) {
                        self.drop_root(old);
                    }
                }
            }
            let first_free = frame.executable.layout().locals.len();
            for (index, cell) in data.closure().iter().enumerate() {
                self.retain(*cell)?;
                frame.localsplus[first_free + index] = Some(Value::Object(*cell));
            }
            Ok(())
        })();
        if let Err(error) = initialized {
            self.clear_frame(frame);
            return Err(error);
        }
        Ok(frame)
    }

    fn invoke_native(
        &mut self,
        function: ObjectRef,
        arguments: &[Value],
    ) -> Result<NativeResult, VmError> {
        self.object(function)?;
        for argument in arguments {
            self.validate_value(argument)?;
        }
        if let Some(Value::Object(receiver)) = arguments.first() {
            self.objects.check_method_contract_call(
                function,
                *receiver,
                self.source_type_checks,
            )?;
        }
        if let Some(arity) = self.objects.call_arity(function)?
            && !arity.accepts(arguments.len())
        {
            return Err(VmError::ArgumentCount {
                name: self.object(function)?.payload().kind_name().into(),
                required: arity.required,
                maximum: arity.maximum,
                provided: arguments.len(),
            });
        }
        if let Some(callable) = self.object(function)?.payload().native_callable() {
            return NativeContext::execute(&mut self.objects, function, arguments, callable);
        }
        let signature = self.objects.native_signature(function)?;
        let name = self.object(function)?.payload().kind_name().to_owned();
        if let Some(signature) = &signature {
            signature.check_arguments(&self.objects, &name, arguments)?;
        }
        let result = self.object(function)?.payload().call(arguments)?;
        if let Some(signature) = &signature {
            signature.check_return(&self.objects, &name, &result)?;
        }
        self.retain_value(&result)?;
        Ok(NativeResult::Return(result))
    }

    fn prepare_call(
        &mut self,
        mut function: ObjectRef,
        mut arguments: Vec<Value>,
        executed: &mut usize,
        limits: ExecutionLimits,
    ) -> Result<PreparedCall, VmError> {
        let mut owned = false;
        let mut returns = Vec::new();
        let result = (|| loop {
            if self
                .object(function)?
                .payload()
                .as_any()
                .is::<EndorFunctionObject>()
            {
                let mut frame = self.prepare_frame(function, &arguments)?;
                let mut local = std::mem::take(&mut frame.native_returns);
                frame.native_returns = std::mem::take(&mut returns);
                frame.native_returns.append(&mut local);
                return Ok(PreparedCall::Frame(frame));
            }
            if *executed >= limits.max_instructions {
                return Err(VmError::ExecutionLimit("instructions"));
            }
            *executed += 1;
            if let Some(signature) = self.objects.native_signature(function)?
                && !signature.returns.is_any()
            {
                let check = NativeReturn {
                    name: self.object(function)?.payload().kind_name().into(),
                    expected: signature.returns,
                };
                self.push_native_return(&mut returns, check)?;
            }
            if let Some(Value::Object(receiver)) = arguments.first() {
                for (name, expected) in self.objects.method_trait_returns(function, *receiver)? {
                    self.push_native_return(&mut returns, NativeReturn { name, expected })?;
                }
            }
            match self.invoke_native(function, &arguments)? {
                NativeResult::Return(value) => {
                    for check in returns.iter().rev() {
                        if let Err(error) = check.check(&self.objects, &value) {
                            self.drop_root(value);
                            return Err(error);
                        }
                    }
                    return Ok(PreparedCall::Value(value));
                }
                NativeResult::Invoke {
                    function: next,
                    arguments: values,
                } => {
                    if owned {
                        self.release(function)?;
                        for value in arguments.drain(..) {
                            self.release_value(value)?;
                        }
                    }
                    function = next;
                    arguments = values;
                    owned = true;
                }
            }
        })();
        self.clear_native_returns(returns);
        if owned {
            self.release(function)
                .expect("delegated call owns its function");
            for value in arguments {
                self.drop_root(value);
            }
        }
        result
    }

    fn push_native_return(
        &mut self,
        returns: &mut Vec<NativeReturn>,
        check: NativeReturn,
    ) -> Result<(), VmError> {
        let mut retained = Vec::new();
        for reference in check.expected.references() {
            if let Err(error) = self.retain(reference) {
                for reference in retained {
                    self.release(reference)?;
                }
                return Err(error);
            }
            retained.push(reference);
        }
        returns.push(check);
        Ok(())
    }
    fn clear_native_returns(&mut self, returns: Vec<NativeReturn>) {
        for check in returns {
            for reference in check.expected.references() {
                self.release(reference)
                    .expect("native return check owns its type");
            }
        }
    }

    fn stack_value(frame: &EndorFrame, index: usize) -> Result<Value, VmError> {
        match frame.stack.get(index) {
            Some(StackEntry::Value(value)) => Ok(value.clone()),
            _ => Err(invalid(
                frame.current_offset,
                "NULL call marker used as a value",
            )),
        }
    }

    fn top_value(frame: &EndorFrame) -> Result<Value, VmError> {
        Self::stack_value(frame, frame.stack.len() - 1)
    }

    fn assignment_value(
        &mut self,
        previous: Option<Value>,
        incoming: Value,
    ) -> Result<Value, VmError> {
        if let (Some(Value::Object(old)), Value::Object(new)) = (&previous, &incoming)
            && (old == new || self.objects.assign_from(*old, *new)?)
        {
            return Ok(Value::Object(*old));
        }
        Ok(incoming)
    }

    fn push_borrowed(&mut self, frame: &mut EndorFrame, value: Value) -> Result<(), VmError> {
        self.retain_value(&value)?;
        frame.stack.push(StackEntry::Value(value));
        Ok(())
    }

    fn discard(&mut self, frame: &mut EndorFrame, count: usize) {
        for _ in 0..count {
            if let StackEntry::Value(value) = frame.stack.pop().expect("verified stack height") {
                self.drop_root(value);
            }
        }
    }

    fn cell_slot(&self, frame: &EndorFrame, index: usize) -> Result<ObjectRef, VmError> {
        match &frame.localsplus[index] {
            Some(Value::Object(reference)) => {
                self.payload::<EndorCellObject>(*reference, "cell")?;
                Ok(*reference)
            }
            _ => Err(invalid(
                frame.current_offset,
                "cell slot is not initialized",
            )),
        }
    }

    fn value_is_true(&self, value: &Value) -> Result<bool, VmError> {
        match value {
            Value::Nil => Ok(false),
            Value::Bool(value) => Ok(*value),
            Value::Text(value) => Ok(!value.is_empty()),
            Value::Number(value) => Ok(!value.is_zero()),
            Value::Object(reference) => self.is_true(*reference),
        }
    }

    fn text_value(&self, value: &Value) -> Result<Option<String>, VmError> {
        Ok(match value {
            Value::Text(text) => Some(text.clone()),
            Value::Object(reference) => self
                .object(*reference)?
                .payload()
                .as_any()
                .downcast_ref::<EndorStringObject>()
                .map(|text| text.as_str().to_owned()),
            _ => None,
        })
    }

    fn number_value<'a>(
        &'a self,
        value: &'a Value,
    ) -> Result<Option<&'a crate::EndorNumber>, VmError> {
        Ok(match value {
            Value::Number(n) => Some(n),
            Value::Object(o) => self
                .object(*o)?
                .payload()
                .as_any()
                .downcast_ref::<crate::EndorNumberObject>()
                .map(|n| n.value()),
            _ => None,
        })
    }

    fn compare_values(
        &self,
        left: &Value,
        right: &Value,
        operation: CompareOp,
    ) -> Result<bool, VmError> {
        let (left_number, right_number) = (self.number_value(left)?, self.number_value(right)?);
        if let (Some(left), Some(right)) = (left_number, right_number) {
            return left.numeric_compare(right, operation);
        }
        if left_number.is_some() || right_number.is_some() {
            return match operation {
                CompareOp::Eq => Ok(false),
                CompareOp::Ne => Ok(true),
                _ => Err(VmError::UnsupportedOperation("mixed numeric ordering")),
            };
        }
        if let (Some(left), Some(right)) = (self.text_value(left)?, self.text_value(right)?) {
            return Ok(match operation {
                CompareOp::Eq => left == right,
                CompareOp::Ne => left != right,
                CompareOp::Lt => left < right,
                CompareOp::Le => left <= right,
                CompareOp::Gt => left > right,
                CompareOp::Ge => left >= right,
            });
        }
        if let (Value::Object(left), Value::Object(right)) = (left, right) {
            return self.compare(*left, *right, operation);
        }
        match operation {
            CompareOp::Eq => Ok(left == right),
            CompareOp::Ne => Ok(left != right),
            _ => Err(VmError::UnsupportedOperation("ordering")),
        }
    }

    fn build_map(&mut self, frame: &EndorFrame, count: usize) -> Result<ObjectRef, VmError> {
        let dictionary = self.new_dict()?;
        let result = (|| -> Result<(), VmError> {
            let start = frame.stack.len() - count * 2;
            for index in 0..count {
                let key = Self::stack_value(frame, start + index * 2)?;
                let value = Self::stack_value(frame, start + index * 2 + 1)?;
                self.dict_set(dictionary, key, value)?;
            }
            Ok(())
        })();
        if let Err(error) = result {
            self.release(dictionary)?;
            return Err(error);
        }
        Ok(dictionary)
    }

    fn make_function(&mut self, frame: &EndorFrame, argument: u32) -> Result<ObjectRef, VmError> {
        let (default_count, closure_count) = Instruction::function_counts(argument);
        let Value::Object(code) = Self::top_value(frame)? else {
            return Err(VmError::ObjectKindMismatch("code"));
        };
        let name = self
            .payload::<EndorCodeObject>(code, "code")?
            .code()
            .name()
            .to_owned();
        let start = frame.stack.len() - default_count - closure_count - 1;
        let defaults = (start..start + default_count)
            .map(|index| Self::stack_value(frame, index))
            .collect::<Result<Vec<_>, _>>()?;
        let closure = (start + default_count..frame.stack.len() - 1)
            .map(|index| match Self::stack_value(frame, index)? {
                Value::Object(cell) => Ok(cell),
                _ => Err(VmError::ObjectKindMismatch("cell")),
            })
            .collect::<Result<Vec<_>, _>>()?;
        // Keep every input root on the stack until allocation has validated
        // and retained all function edges. Failure leaves cleanup to the frame.
        self.new_function(name, code, frame.globals, defaults, closure)
    }

    fn build_class(&mut self, frame: &EndorFrame) -> Result<ObjectRef, VmError> {
        let start = frame.stack.len() - 4;
        let Value::Text(name) = Self::stack_value(frame, start)? else {
            return Err(VmError::ObjectKindMismatch("class name"));
        };
        let Value::Object(base) = Self::stack_value(frame, start + 1)? else {
            return Err(VmError::ObjectKindMismatch("base Type"));
        };
        let Value::Object(fields) = Self::stack_value(frame, start + 2)? else {
            return Err(VmError::ObjectKindMismatch("field map"));
        };
        let Value::Object(methods) = Self::stack_value(frame, start + 3)? else {
            return Err(VmError::ObjectKindMismatch("method map"));
        };
        let fields = self
            .dict_items(fields)?
            .iter()
            .map(|(name, type_name)| match (name, type_name) {
                (Value::Text(name), Value::Text(type_name)) => Ok(crate::ClassField {
                    name: name.clone(),
                    type_name: type_name.clone(),
                }),
                _ => Err(VmError::ObjectKindMismatch("class field annotation")),
            })
            .collect::<Result<Vec<_>, _>>()?;
        let methods = self
            .dict_items(methods)?
            .iter()
            .map(|(name, function)| match (name, function) {
                (Value::Text(name), Value::Object(function)) => Ok((name.clone(), *function)),
                _ => Err(VmError::ObjectKindMismatch("class method")),
            })
            .collect::<Result<std::collections::BTreeMap<_, _>, _>>()?;
        self.objects
            .new_class_with_scope(name, base, fields, methods, Some(frame.function))
    }

    fn eval_frames(
        &mut self,
        frames: &mut Vec<EndorFrame>,
        limits: ExecutionLimits,
        executed: &mut usize,
        slice_end: usize,
        allow_wait: bool,
    ) -> Result<Option<Value>, VmError> {
        loop {
            if *executed >= limits.max_instructions {
                return Err(VmError::ExecutionLimit("instructions"));
            }
            if *executed >= slice_end {
                return Ok(None);
            }
            *executed += 1;
            let frame = frames.last_mut().expect("an evaluation frame is active");
            let decoded = frame.executable.instruction_at(frame.instruction_offset)?;
            frame.current_offset = decoded.offset;
            frame.instruction_offset = decoded.next_offset;
            let opcode = decoded.instruction.opcode;
            let argument = decoded.instruction.argument as usize;
            match opcode {
                Opcode::Await => {
                    if self.objects.is_read_only() {
                        return Err(VmError::UnsupportedOperation(
                            "await during read-only execution",
                        ));
                    }
                    let Value::Object(awaitable) = Self::top_value(frame)? else {
                        return Err(VmError::ObjectKindMismatch("awaitable object"));
                    };
                    match self.object(awaitable)?.payload().poll_await()? {
                        crate::AwaitPoll::Pending => {
                            if !allow_wait {
                                return Err(VmError::UnsupportedOperation(
                                    "pending await outside a coroutine",
                                ));
                            }
                            // Keep the evaluated operand, never replay its call
                            // or the statements before this await instruction.
                            frame.waiting = true;
                            frame.instruction_offset = decoded.offset;
                            return Ok(None);
                        }
                        crate::AwaitPoll::Ready(value) => {
                            self.retain_value(&value)?;
                            self.discard(frame, 1);
                            frame.stack.push(StackEntry::Value(value));
                            frame.waiting = false;
                        }
                    }
                }
                Opcode::Nop => {}
                Opcode::LoadConst => {
                    let value = frame.executable.constants()[argument].clone();
                    self.push_borrowed(frame, value)?;
                }
                Opcode::LoadFast => {
                    let value = frame.localsplus[argument].clone().ok_or_else(|| {
                        VmError::UnboundLocal(frame.executable.slot_name(argument).into())
                    })?;
                    self.push_borrowed(frame, value)?;
                }
                Opcode::StoreFast => {
                    let value = self.assignment_value(
                        frame.localsplus[argument].clone(),
                        Self::top_value(frame)?,
                    )?;
                    self.retain_value(&value)?;
                    self.discard(frame, 1);
                    if let Some(old) = frame.localsplus[argument].replace(value) {
                        self.drop_root(old);
                    }
                }
                Opcode::DeleteFast => {
                    let value = frame.localsplus[argument].take().ok_or_else(|| {
                        VmError::UnboundLocal(frame.executable.slot_name(argument).into())
                    })?;
                    self.drop_root(value);
                }
                Opcode::LoadGlobal => {
                    let name = &frame.executable.layout().names[argument];
                    let mut value = self.objects.dict_get_name(frame.globals, name)?;
                    if value.is_none()
                        && let Some(builtins) = frame.builtins
                    {
                        value = self.objects.dict_get_name(builtins, name)?;
                    }
                    let value = value.ok_or_else(|| VmError::NameNotFound(name.clone()))?;
                    self.push_borrowed(frame, value)?;
                }
                Opcode::StoreGlobal => {
                    let key = Value::Text(frame.executable.layout().names[argument].clone());
                    let value = self.assignment_value(
                        self.dict_get(frame.globals, &key)?,
                        Self::top_value(frame)?,
                    )?;
                    self.dict_set(frame.globals, key, value)?;
                    self.discard(frame, 1);
                }
                Opcode::DeleteGlobal => {
                    let name = &frame.executable.layout().names[argument];
                    if !self.dict_delete(frame.globals, &Value::Text(name.clone()))? {
                        return Err(VmError::NameNotFound(name.clone()));
                    }
                }
                Opcode::LoadDeref | Opcode::LoadClosure => {
                    let cell = self.cell_slot(frame, argument)?;
                    let value = if opcode == Opcode::LoadClosure {
                        Value::Object(cell)
                    } else {
                        self.cell_get(cell).map_err(|error| {
                            if error == VmError::EmptyCell {
                                VmError::UnboundLocal(frame.executable.slot_name(argument).into())
                            } else {
                                error
                            }
                        })?
                    };
                    self.push_borrowed(frame, value)?;
                }
                Opcode::StoreDeref => {
                    let cell = self.cell_slot(frame, argument)?;
                    let value =
                        self.assignment_value(self.cell_get(cell).ok(), Self::top_value(frame)?)?;
                    self.cell_set(cell, value)?;
                    self.discard(frame, 1);
                }
                Opcode::DeleteDeref => {
                    let cell = self.cell_slot(frame, argument)?;
                    if self.cell_get(cell).is_err() {
                        return Err(VmError::UnboundLocal(
                            frame.executable.slot_name(argument).into(),
                        ));
                    }
                    self.cell_clear(cell)?;
                }
                Opcode::PopTop => self.discard(frame, 1),
                Opcode::Copy => {
                    let entry = frame.stack[frame.stack.len() - argument].clone();
                    if let StackEntry::Value(value) = &entry {
                        self.retain_value(value)?;
                    }
                    frame.stack.push(entry);
                }
                Opcode::Swap => {
                    let length = frame.stack.len();
                    frame.stack.swap(length - 1, length - argument);
                }
                Opcode::PushNull => frame.stack.push(StackEntry::Null),
                Opcode::Call => {
                    let start = frame.stack.len() - argument - 2;
                    let Value::Object(function) = Self::stack_value(frame, start)? else {
                        return Err(VmError::UnsupportedOperation("call"));
                    };
                    let mut arguments = Vec::with_capacity(argument + 1);
                    if let StackEntry::Value(value) = &frame.stack[start + 1] {
                        arguments.push(value.clone());
                    }
                    for index in start + 2..frame.stack.len() {
                        arguments.push(Self::stack_value(frame, index)?);
                    }
                    match self.prepare_call(function, arguments, executed, limits)? {
                        PreparedCall::Frame(mut callee) => {
                            if frames.len() >= limits.max_frames {
                                self.clear_frame(callee);
                                return Err(VmError::ExecutionLimit("frames"));
                            }
                            callee.previous = Some(frames.len() - 1);
                            self.discard(frames.last_mut().unwrap(), argument + 2);
                            frames.push(callee);
                        }
                        PreparedCall::Value(value) => {
                            self.discard(frames.last_mut().unwrap(), argument + 2);
                            frames
                                .last_mut()
                                .unwrap()
                                .stack
                                .push(StackEntry::Value(value));
                        }
                    }
                }
                Opcode::ReturnValue => {
                    let result = Self::top_value(frame)?;
                    if let Some(expected) = frame.return_type
                        && !self.objects.matches_annotation(&result, expected)?
                    {
                        return Err(VmError::TypeMismatch {
                            function: frame.name.clone(),
                            position: "return".into(),
                            expected: frame
                                .executable
                                .source_signature()
                                .expect("checked source signature")
                                .return_type
                                .clone(),
                            actual: self.objects.value_type_name(&result)?,
                        });
                    }
                    for check in frame.native_returns.iter().rev() {
                        check.check(&self.objects, &result)?;
                    }
                    frame.stack.pop();
                    self.clear_frame(frames.pop().unwrap());
                    if let Some(caller) = frames.last_mut() {
                        caller.stack.push(StackEntry::Value(result));
                    } else {
                        return Ok(Some(result));
                    }
                }
                Opcode::JumpForward | Opcode::JumpBackward => {
                    frame.instruction_offset = decoded.jump_target()?
                }
                Opcode::PopJumpIfFalse | Opcode::PopJumpIfTrue => {
                    let truth = self.value_is_true(&Self::top_value(frame)?)?;
                    self.discard(frame, 1);
                    if truth == (opcode == Opcode::PopJumpIfTrue) {
                        frame.instruction_offset = decoded.jump_target()?;
                    }
                }
                Opcode::ToBool | Opcode::UnaryNot => {
                    let truth = self.value_is_true(&Self::top_value(frame)?)?;
                    self.discard(frame, 1);
                    frame.stack.push(StackEntry::Value(Value::Bool(
                        if opcode == Opcode::UnaryNot {
                            !truth
                        } else {
                            truth
                        },
                    )));
                }
                Opcode::BinaryArithmetic => {
                    if decoded.instruction.argument == 0 {
                        let left = Self::stack_value(frame, frame.stack.len() - 2)?;
                        let right = Self::top_value(frame)?;
                        if let (Value::Text(a), Value::Text(b)) = (left, right) {
                            if a.len().saturating_add(b.len()) > 1_048_576 {
                                return Err(VmError::InvalidArgument(
                                    "text concatenation exceeds operation budget",
                                ));
                            }
                            self.discard(frame, 2);
                            frame.stack.push(StackEntry::Value(Value::Text(a + &b)));
                            continue;
                        }
                    }
                    let left = Self::stack_value(frame, frame.stack.len() - 2)?;
                    let right = Self::top_value(frame)?;
                    let left = self
                        .number_value(&left)?
                        .ok_or(VmError::ObjectKindMismatch("Number"))?;
                    let right = self
                        .number_value(&right)?
                        .ok_or(VmError::ObjectKindMismatch("Number"))?;
                    let result = left.arithmetic(
                        right,
                        crate::ArithmeticOp::from_argument(decoded.instruction.argument).unwrap(),
                    )?;
                    self.discard(frame, 2);
                    frame.stack.push(StackEntry::Value(Value::Number(result)));
                }
                Opcode::UnaryNegate => {
                    let value = Self::top_value(frame)?;
                    let value = self
                        .number_value(&value)?
                        .ok_or(VmError::ObjectKindMismatch("Number"))?;
                    let result = value.negated()?;
                    self.discard(frame, 1);
                    frame.stack.push(StackEntry::Value(Value::Number(result)));
                }
                Opcode::CompareOp => {
                    let left = Self::stack_value(frame, frame.stack.len() - 2)?;
                    let right = Self::top_value(frame)?;
                    let result = self.compare_values(
                        &left,
                        &right,
                        Instruction::compare_operation(decoded.instruction.argument).unwrap(),
                    )?;
                    self.discard(frame, 2);
                    frame.stack.push(StackEntry::Value(Value::Bool(result)));
                }
                Opcode::LoadAttr => {
                    let Value::Object(owner) = Self::top_value(frame)? else {
                        return Err(VmError::ObjectKindMismatch("object"));
                    };
                    let value =
                        self.bind_attr(owner, &frame.executable.layout().names[argument])?;
                    self.discard(frame, 1);
                    frame.stack.push(StackEntry::Value(value));
                }
                Opcode::BuildClass => {
                    let class = self.build_class(frame)?;
                    self.discard(frame, 4);
                    frame.stack.push(StackEntry::Value(Value::Object(class)));
                }
                Opcode::StoreAttr => {
                    let Value::Object(owner) = Self::top_value(frame)? else {
                        return Err(VmError::ObjectKindMismatch("object"));
                    };
                    let value = Self::stack_value(frame, frame.stack.len() - 2)?;
                    self.set_attr(
                        owner,
                        frame.executable.layout().names[argument].clone(),
                        value,
                    )?;
                    self.discard(frame, 2);
                }
                Opcode::BuildMap => {
                    let dictionary = self.build_map(frame, argument)?;
                    self.discard(frame, argument * 2);
                    frame
                        .stack
                        .push(StackEntry::Value(Value::Object(dictionary)));
                }
                Opcode::MakeFunction => {
                    let function = self.make_function(frame, decoded.instruction.argument)?;
                    let (defaults, closure) =
                        Instruction::function_counts(decoded.instruction.argument);
                    self.discard(frame, defaults + closure + 1);
                    frame.stack.push(StackEntry::Value(Value::Object(function)));
                }
                Opcode::ExtendedArg => unreachable!("EXTENDED_ARG was decoded before evaluation"),
            }
        }
    }
}