//! Bottom-up measurement and top-down arrangement share one read-only VM call.
//! The host scratch plan contains derived data only, never a second business heap.
use crate::{
    document::Frame,
    layout::{self, LayoutRuntime, Size},
    mount::Mount,
    native_visual::{number, record, value},
    object_list,
    ui_bridge::{MAX_DIMENSION, UiNode, desktop_type},
};
use anyhow::Result;
use endor_core::{
    ContextNativeFunction, NativeCallable, NativeContext, NativeResult, ObjectRef, Value, Vm,
    VmError, WeakObjectRef,
};
use std::{
    collections::{BTreeSet, HashMap, HashSet},
    sync::{Arc, Mutex},
};

// Small native trees keep the allocation-free fast path. Larger trees use the
// same per-occurrence VM driver so even intrinsic work can yield between nodes.
const NATIVE_FAST_PATH_MAX_NODES: usize = 64;

fn invalid(message: &'static str) -> VmError {
    VmError::InvalidArgument(message)
}
fn desktop_type_registry(cx: &NativeContext<'_>, desktop: ObjectRef) -> Result<ObjectRef, VmError> {
    match cx.store().get_attr(desktop, "types")? {
        Value::Object(object) => Ok(object),
        _ => Err(invalid("missing type registry")),
    }
}
#[derive(Debug, Clone)]
struct Measurement {
    owner: ObjectRef,
    method: ObjectRef,
    children: Vec<Size>,
    constraints: [f32; 4],
    size: Size,
    reads: Vec<endor_core::ReadDependency>,
}
/// Borrowed identities and versions only: this cache adds no managed roots.
#[derive(Debug, Default, Clone)]
pub struct Cache {
    entries: HashMap<WeakObjectRef, Measurement>,
    pub hits: usize,
    pub misses: usize,
}
impl Cache {
    pub(crate) fn exclude_host_desktops(&mut self, desktops: &HashSet<ObjectRef>) {
        for entry in self.entries.values_mut() {
            entry
                .reads
                .retain(|read| !desktops.contains(&read.object()));
        }
    }
    pub fn invalidate(&mut self, owners: &HashSet<ObjectRef>) {
        self.entries
            .retain(|_, entry| !owners.contains(&entry.owner));
    }
    pub fn clear(&mut self) {
        self.entries.clear();
        self.hits = 0;
        self.misses = 0;
    }
}
#[derive(Debug)]
struct Pending {
    token: usize,
    owner: ObjectRef,
    method: ObjectRef,
    children: Vec<Size>,
    constraints: [f32; 4],
    key: WeakObjectRef,
}
#[derive(Debug)]
struct Plan {
    nodes: Vec<Arc<UiNode>>,
    children: Vec<Vec<usize>>,
    sizes: Vec<Size>,
    preferred: Vec<Option<Size>>,
    bounds: Vec<Option<[f32; 4]>>,
    runtime: LayoutRuntime,
    custom: bool,
    cache: Cache,
    pending: Vec<Option<Pending>>,
    constraints: Vec<[f32; 4]>,
    // Source-arranged dimensions remain exact for this solve, not business state.
    allocations: Vec<[Option<f32>; 2]>,
    custom_arrange: Vec<bool>,
    native_arrange: Vec<bool>,
    custom_measure: Vec<bool>,
    measured_limits: Vec<[Option<f32>; 2]>,
    previous: Option<Vec<u32>>,
    seen: BTreeSet<Vec<u32>>,
}
type Shared = Arc<Mutex<Plan>>;
#[derive(Debug, Clone, Copy)]
enum Phase {
    Measure,
    Arrange,
}
impl Phase {
    fn method(self) -> &'static str {
        match self {
            Self::Measure => "measure",
            Self::Arrange => "arrange",
        }
    }
}
#[derive(Debug)]
struct Step {
    plan: Shared,
    index: usize,
    phase: Phase,
    mount: ObjectRef,
    desktop: ObjectRef,
}
impl NativeCallable for Step {
    fn arity(&self) -> Option<endor_core::CallArity> {
        Some(endor_core::CallArity::exact(0))
    }

    fn invoke(&self, cx: &mut NativeContext<'_>, _: &[Value]) -> Result<NativeResult, VmError> {
        let mount = cx.store().payload::<Mount>(self.mount, "Mount")?.clone();
        let definition = cx
            .store()
            .type_of(mount.block)?
            .ok_or(invalid("untyped layout target"))?;
        let method = cx
            .store()
            .lookup_method(definition, self.phase.method())?
            .ok_or(invalid("missing layout method"))?;
        let intrinsic = crate::native_visual::is_intrinsic(
            cx.store(),
            method,
            self.phase.method(),
            self.desktop,
        )?;
        let (node, children, preferred, bounds, native_size) = {
            let mut plan = self
                .plan
                .lock()
                .map_err(|_| invalid("layout plan poisoned"))?;
            let node = plan.nodes[self.index].clone();
            let children = plan.children[self.index]
                .iter()
                .map(|child| plan.sizes[*child])
                .collect::<Vec<_>>();
            let preferred = plan.children[self.index]
                .iter()
                .map(|child| plan.preferred[*child].unwrap_or(plan.sizes[*child]))
                .collect::<Vec<_>>();
            let bounds = if matches!(self.phase, Phase::Arrange) {
                plan.bounds[self.index]
            } else {
                None
            };
            if matches!(self.phase, Phase::Measure) {
                plan.custom_measure[self.index] = !intrinsic;
            }
            let native_size = if matches!(self.phase, Phase::Measure) && intrinsic {
                let constraints = plan.constraints[self.index];
                Some(
                    plan.runtime
                        .measure_node(&node, children.clone(), constraints)
                        .map_err(|error| {
                            error.downcast_ref::<VmError>().cloned().unwrap_or_else(|| {
                                invalid("native measurement exceeds layout limits")
                            })
                        })?,
                )
            } else {
                if !intrinsic {
                    plan.custom = true;
                }
                if matches!(self.phase, Phase::Measure) {
                    let key = cx.store().downgrade(self.mount)?;
                    let constraints = plan.constraints[self.index];
                    let cached =
                        plan.cache
                            .entries
                            .get(&key)
                            .filter(|entry| {
                                entry.owner == mount.block
                                    && entry.method == method
                                    && entry.children == children
                                    && entry.constraints == constraints
                                    && entry.reads.iter().all(|dependency| {
                                        cx.store().read_dependency_valid(dependency)
                                    })
                            })
                            .map(|entry| entry.size);
                    if let Some(size) = cached {
                        plan.cache.hits += 1;
                        Some(size)
                    } else {
                        plan.cache.misses += 1;
                        plan.runtime.measured += 1;
                        plan.pending[self.index] = Some(Pending {
                            token: 0,
                            owner: mount.block,
                            method,
                            children: children.clone(),
                            constraints,
                            key,
                        });
                        None
                    }
                } else {
                    None
                }
            };
            if matches!(self.phase, Phase::Arrange) {
                plan.runtime.arranged += 1;
                plan.native_arrange[self.index] = intrinsic;
                plan.custom_arrange[self.index] = !intrinsic || node.allocates_children();
            }
            (node, children, preferred, bounds, native_size)
        };
        let context = mount.layout.ok_or(invalid("missing pre-layout context"))?;
        {
            let mut context = context
                .lock()
                .map_err(|_| invalid("layout context poisoned"))?;
            context.node = Some(node);
            context.children = children;
            context.preferred = preferred;
            context.bounds = bounds;
        }
        if let Some(size) = native_size {
            return Ok(NativeResult::Return(Value::Object(record(
                cx,
                self.desktop,
                "MeasureResult",
                vec![("width", value(size.width)), ("height", value(size.height))],
            )?)));
        }
        let input = match self.phase {
            Phase::Measure => {
                let constraints = self
                    .plan
                    .lock()
                    .map_err(|_| invalid("layout plan poisoned"))?
                    .constraints[self.index];
                record(
                    cx,
                    self.desktop,
                    "Constraints",
                    vec![
                        ("minWidth", value(constraints[0])),
                        ("minHeight", value(constraints[1])),
                        ("maxWidth", value(constraints[2])),
                        ("maxHeight", value(constraints[3])),
                    ],
                )?
            }
            Phase::Arrange => {
                let [x, y, width, height] =
                    bounds.ok_or(invalid("arrangement has no allocated bounds"))?;
                record(
                    cx,
                    self.desktop,
                    "Bounds",
                    vec![
                        ("x", value(x)),
                        ("y", value(y)),
                        ("width", value(width)),
                        ("height", value(height)),
                    ],
                )?
            }
        };
        if matches!(self.phase, Phase::Measure) {
            let token = cx.store().begin_read_tracking()?;
            self.plan
                .lock()
                .map_err(|_| invalid("layout plan poisoned"))?
                .pending[self.index]
                .as_mut()
                .ok_or(invalid("missing measurement tracking state"))?
                .token = token;
        }
        Ok(NativeResult::Invoke {
            function: method,
            arguments: vec![
                Value::Object(mount.block),
                Value::Object(self.mount),
                Value::Object(input),
            ],
        })
    }
}
#[derive(Debug)]
struct Collect {
    plan: Shared,
    index: usize,
    phase: Phase,
    desktop: ObjectRef,
}
impl NativeCallable for Collect {
    fn arity(&self) -> Option<endor_core::CallArity> {
        Some(endor_core::CallArity::exact(1))
    }

    fn invoke(&self, cx: &mut NativeContext<'_>, args: &[Value]) -> Result<NativeResult, VmError> {
        let pending = if matches!(self.phase, Phase::Measure) {
            self.plan
                .lock()
                .map_err(|_| invalid("layout plan poisoned"))?
                .pending[self.index]
                .take()
        } else {
            None
        };
        let [Value::Object(result)] = args else {
            return Err(invalid("layout must return a managed result"));
        };
        let expected = match self.phase {
            Phase::Measure => "MeasureResult",
            Phase::Arrange => "Arrangement",
        };
        if !cx
            .store()
            .is_instance(*result, desktop_type(cx.store(), expected, self.desktop)?)?
        {
            return Err(invalid("layout returned the wrong result type"));
        }
        let mut plan = self
            .plan
            .lock()
            .map_err(|_| invalid("layout plan poisoned"))?;
        match self.phase {
            Phase::Measure => {
                let size = layout::constrained(
                    Size {
                        width: number(cx, *result, "width", false)?,
                        height: number(cx, *result, "height", false)?,
                    },
                    plan.constraints[self.index],
                );
                // Native shrink-to-content containers keep their existing input
                // policy. Only a source measurement that differs from the native
                // result supplies an additional content constraint.
                if plan.custom_measure[self.index] && !plan.children[self.index].is_empty() {
                    let children = plan.children[self.index]
                        .iter()
                        .map(|child| plan.sizes[*child])
                        .collect();
                    let native = layout::measure_native(
                        &plan.nodes[self.index],
                        children,
                        plan.constraints[self.index],
                    )
                    .ok()
                    .map(|native| layout::constrained(native, plan.constraints[self.index]));
                    // Equality after remeasurement does not retract a source
                    // limit selected earlier in this solve. New renders start fresh.
                    if !native.is_some_and(|native| native.width == size.width) {
                        plan.measured_limits[self.index][0] = Some(size.width);
                    }
                    if !native.is_some_and(|native| native.height == size.height) {
                        plan.measured_limits[self.index][1] = Some(size.height);
                    }
                }
                // Decoding is part of the subcall dependency region: a source
                // method may return an existing result whose fields can change.
                let tracked = pending
                    .map(|pending| {
                        cx.store()
                            .end_read_tracking_precise(pending.token)
                            .map(|reads| (pending, reads))
                    })
                    .transpose()?;
                if let Some((pending, reads)) = tracked {
                    let mut volatile = false;
                    let registry = desktop_type_registry(cx, self.desktop)?;
                    for read in &reads.accesses {
                        let object = read.object;
                        if object == self.desktop || object == registry {
                            continue;
                        }
                        // A Mount's layoutVersion/geometry may be observed by
                        // source. Object-granular tracking cannot prove those
                        // accesses stable, so conservatively bypass its cache.
                        if cx.store().object(object)?.payload().as_any().is::<Mount>() {
                            volatile |= read.content_read;
                            continue;
                        }
                    }
                    if volatile {
                        plan.cache.entries.remove(&pending.key);
                    } else {
                        plan.cache.entries.insert(
                            pending.key,
                            Measurement {
                                owner: pending.owner,
                                method: pending.method,
                                children: pending.children,
                                constraints: pending.constraints,
                                size,
                                reads: reads
                                    .dependencies
                                    .into_iter()
                                    .filter(|dependency| {
                                        let object = dependency.object();
                                        object != self.desktop
                                            && object != registry
                                            && cx.store().object(object).is_ok_and(|value| {
                                                !value.payload().as_any().is::<Mount>()
                                            })
                                    })
                                    .collect(),
                            },
                        );
                    }
                }
                plan.preferred[self.index].get_or_insert(size);
                plan.sizes[self.index] = size;
                if self.index == 0 {
                    plan.bounds[0] = Some([0.0, 0.0, size.width, size.height]);
                }
            }
            Phase::Arrange => {
                let Value::Object(list) = cx.store().get_attr(*result, "children")? else {
                    return Err(invalid("Arrangement.children must be an ObjectList"));
                };
                let items = object_list::items(cx.store(), list)?;
                let children = plan.children[self.index].clone();
                if items.len() != children.len() {
                    return Err(invalid(
                        "Arrangement must assign every child slot exactly once",
                    ));
                }
                let slots: HashMap<_, _> = plan.nodes[self.index]
                    .slots
                    .iter()
                    .copied()
                    .enumerate()
                    .map(|(index, slot)| (slot, index))
                    .collect();
                let mut assigned = BTreeSet::new();
                let expected = desktop_type(cx.store(), "Bounds", self.desktop)?;
                for item in items {
                    if !cx.store().is_instance(*item, expected)? {
                        return Err(invalid("Arrangement child must be Bounds"));
                    }
                    let Value::Number(slot) = cx.store().get_attr(*item, "slot")? else {
                        return Err(invalid("Arrangement child requires a slot"));
                    };
                    let slot = slot
                        .as_i64()
                        .and_then(|value| u64::try_from(value).ok())
                        .ok_or(invalid("invalid arrangement slot"))?;
                    let position = slots
                        .get(&slot)
                        .copied()
                        .ok_or(invalid("Arrangement references a foreign child slot"))?;
                    if !assigned.insert(slot) {
                        return Err(invalid("duplicate arrangement slot"));
                    }
                    let bounds = [
                        number(cx, *item, "x", true)?,
                        number(cx, *item, "y", true)?,
                        number(cx, *item, "width", false)?,
                        number(cx, *item, "height", false)?,
                    ];
                    if bounds[0] + bounds[2] > MAX_DIMENSION as f32
                        || bounds[1] + bounds[3] > MAX_DIMENSION as f32
                    {
                        return Err(invalid("arrangement bounds exceed limits"));
                    }
                    let child = children[position];
                    if plan.custom_arrange[self.index] {
                        if plan.native_arrange[self.index]
                            && let Some(main) = layout::main_axis(&plan.nodes[self.index])
                        {
                            // Keep an allocation explicit even if this pass's
                            // natural size happens to match it. It must survive
                            // the next parent-constraint propagation pass.
                            plan.allocations[child][main] = Some(bounds[main + 2]);
                        }
                        if bounds[2] != plan.sizes[child].width {
                            plan.allocations[child][0] = Some(bounds[2]);
                        }
                        if bounds[3] != plan.sizes[child].height {
                            plan.allocations[child][1] = Some(bounds[3]);
                        }
                    }
                    plan.bounds[child] = Some(bounds);
                }
            }
        }
        Ok(NativeResult::Return(Value::Nil))
    }
}
#[derive(Debug)]
struct Settle {
    plan: Shared,
    mounts: Arc<[WeakObjectRef]>,
}
impl NativeCallable for Settle {
    fn arity(&self) -> Option<endor_core::CallArity> {
        Some(endor_core::CallArity::exact(0))
    }
    fn invoke(&self, cx: &mut NativeContext<'_>, _: &[Value]) -> Result<NativeResult, VmError> {
        let repeat = {
            let mut plan = self
                .plan
                .lock()
                .map_err(|_| invalid("layout plan poisoned"))?;
            plan.runtime.settle_passes += 1;
            let mut next = plan.constraints.clone();
            for index in 0..plan.nodes.len() {
                let children = plan.children[index].clone();
                let [_, _, width, height] =
                    plan.bounds[index].ok_or(invalid("unarranged occurrence"))?;
                let mut available = next[index];
                if let Some(width) = plan.measured_limits[index][0] {
                    available[2] = width;
                }
                if let Some(height) = plan.measured_limits[index][1] {
                    available[3] = height;
                }
                if width != plan.sizes[index].width {
                    available[2] = width;
                }
                if height != plan.sizes[index].height {
                    available[3] = height;
                }
                let base = layout::arranged_child_constraints(
                    &plan.nodes[index],
                    available,
                    Size { width, height },
                );
                for (child_index, child) in children.into_iter().enumerate() {
                    let natural = layout::occurrence_constraints(
                        &plan.nodes[index],
                        &plan.nodes[child],
                        base,
                        Size { width, height },
                        child_index,
                    );
                    let mut constraints = natural;
                    if let Some(width) = plan.allocations[child][0] {
                        constraints[0] = width;
                        constraints[2] = width;
                    }
                    if let Some(height) = plan.allocations[child][1] {
                        constraints[1] = height;
                        constraints[3] = height;
                    }
                    if let Some(main) = layout::main_axis(&plan.nodes[index]) {
                        let cross = 1 - main;
                        if constraints[cross] != plan.constraints[child][cross]
                            || constraints[cross + 2] != plan.constraints[child][cross + 2]
                        {
                            // Wrapping or other cross-dependent measurement
                            // invalidates this occurrence's natural main size.
                            plan.preferred[child] = None;
                            if plan.native_arrange[index] {
                                // Derived flex sizes must not constrain their
                                // own replacement measurement. Source arrange
                                // allocations remain authoritative.
                                plan.allocations[child][main] = None;
                                constraints[main] = natural[main];
                                constraints[main + 2] = natural[main + 2];
                            }
                        }
                    }
                    next[child] = constraints;
                }
            }
            let mut state = Vec::with_capacity(plan.nodes.len() * 14);
            let bits = |value: f32| if value == 0.0 { 0 } else { value.to_bits() };
            for (index, constraints) in next.iter().enumerate() {
                state.extend(constraints.iter().copied().map(bits));
                state.extend([
                    bits(plan.sizes[index].width),
                    bits(plan.sizes[index].height),
                ]);
                state.extend(
                    plan.bounds[index]
                        .ok_or(invalid("unarranged occurrence"))?
                        .into_iter()
                        .map(bits),
                );
                for dimension in plan.allocations[index] {
                    state.push(u32::from(dimension.is_some()));
                    state.push(dimension.map_or(0, bits));
                }
                state.push(u32::from(plan.preferred[index].is_some()));
                if let Some(size) = plan.preferred[index] {
                    state.extend([bits(size.width), bits(size.height)]);
                } else {
                    state.extend([0, 0]);
                }
            }
            if plan.previous.as_ref() == Some(&state) && plan.constraints == next {
                false
            } else {
                if !plan.seen.insert(state.clone()) {
                    return Err(invalid("layout constraints did not converge (cycle)"));
                }
                plan.previous = Some(state);
                plan.constraints = next;
                plan.bounds.fill(None);
                true
            }
        };
        if repeat {
            // No stale geometry from a prior pass is observable during measurement.
            for weak in self.mounts.iter() {
                let mount = cx.upgrade(*weak)?.ok_or(invalid("layout Mount expired"))?;
                if let Some(context) = &cx.store().payload::<Mount>(mount, "Mount")?.layout {
                    context
                        .lock()
                        .map_err(|_| invalid("layout context poisoned"))?
                        .bounds = None;
                }
            }
        }
        Ok(NativeResult::Return(Value::Bool(repeat)))
    }
}

#[derive(Debug)]
struct Prepare {
    plan: Shared,
    mounts: Vec<WeakObjectRef>,
    desktop: ObjectRef,
}
impl NativeCallable for Prepare {
    fn arity(&self) -> Option<endor_core::CallArity> {
        Some(endor_core::CallArity::exact(0))
    }
    fn invoke(&self, cx: &mut NativeContext<'_>, _: &[Value]) -> Result<NativeResult, VmError> {
        let advance = cx.allocate(ContextNativeFunction::new(
            "next layout occurrence",
            Advance {
                plan: self.plan.clone(),
                mounts: self.mounts.clone().into(),
                desktop: self.desktop,
                next: Mutex::new(0),
            },
            vec![self.desktop],
        ))?;
        let function = crate::vm_sequence::repeat_call(cx, advance, "source layout", self.desktop)?;
        Ok(NativeResult::Invoke {
            function,
            arguments: vec![],
        })
    }
}

#[derive(Debug)]
struct Advance {
    plan: Shared,
    mounts: Arc<[WeakObjectRef]>,
    desktop: ObjectRef,
    next: Mutex<usize>,
}
impl NativeCallable for Advance {
    fn arity(&self) -> Option<endor_core::CallArity> {
        Some(endor_core::CallArity::exact(0))
    }
    fn invoke(&self, cx: &mut NativeContext<'_>, _: &[Value]) -> Result<NativeResult, VmError> {
        let count = self.mounts.len();
        let position = {
            let mut next = self
                .next
                .lock()
                .map_err(|_| invalid("layout cursor poisoned"))?;
            let position = *next;
            if position < count * 2 {
                *next += 1;
            }
            position
        };
        if position == count * 2 {
            let settled = Settle {
                plan: self.plan.clone(),
                mounts: self.mounts.clone(),
            }
            .invoke(cx, &[])?;
            let NativeResult::Return(Value::Bool(repeat)) = settled else {
                return Err(invalid("invalid settling result"));
            };
            if repeat {
                *self
                    .next
                    .lock()
                    .map_err(|_| invalid("layout cursor poisoned"))? = 0;
            }
            return Ok(NativeResult::Return(Value::Bool(repeat)));
        }
        let (phase, index) = if position < count {
            (Phase::Measure, count - position - 1)
        } else {
            (Phase::Arrange, position - count)
        };
        let mount = cx
            .upgrade(self.mounts[index])?
            .ok_or(invalid("layout Mount expired"))?;
        let step = cx.allocate(ContextNativeFunction::new(
            "layout step",
            Step {
                plan: self.plan.clone(),
                index,
                phase,
                mount,
                desktop: self.desktop,
            },
            vec![mount, self.desktop],
        ))?;
        let collector = cx.allocate(ContextNativeFunction::new(
            "collect layout",
            Collect {
                plan: self.plan.clone(),
                index,
                phase,
                desktop: self.desktop,
            },
            vec![self.desktop],
        ))?;
        let function = crate::vm_sequence::collect_and_continue(
            cx,
            collector,
            step,
            vec![],
            "layout occurrence",
            self.desktop,
        )?;
        Ok(NativeResult::Invoke {
            function,
            arguments: vec![],
        })
    }
}

/// An unpublished layout frame and derived caches. These are merged only after
/// the driver completes; they are not another copy of application state.
pub struct Draft {
    pub frame: Frame,
    pub runtime: LayoutRuntime,
    pub cache: Cache,
    pub reads: Vec<endor_core::ReadDependency>,
}

#[must_use = "pending layout work must be resumed or cancelled with its owning VM"]
pub enum Progress {
    Pending(Work),
    Complete(Draft),
}

/// Host-owned continuation. No borrowed frame/cache references cross a pause.
/// The prepare handle owns one root until completion or explicit cancellation.
pub struct Work {
    state: Draft,
    plan: Shared,
    prepare: ObjectRef,
    desktop: ObjectRef,
    slices: usize,
}

impl Work {
    pub fn resume(mut self, vm: &mut Vm, slice: usize) -> Result<Progress> {
        self.slices += 1;
        let progress = vm.resume_read_only_call(slice);
        self.accept(vm, progress)
    }

    pub fn cancel(self, vm: &mut Vm) -> Result<()> {
        let cancelled = vm.cancel_read_only_call();
        let released = vm.release(self.prepare);
        cancelled?;
        released?;
        Ok(())
    }

    fn accept(
        self,
        vm: &mut Vm,
        progress: Result<endor_core::ExecutionProgress, VmError>,
    ) -> Result<Progress> {
        match progress {
            Ok(endor_core::ExecutionProgress::Pending { .. }) => Ok(Progress::Pending(self)),
            Ok(endor_core::ExecutionProgress::Complete {
                value,
                dependencies,
                ..
            }) => {
                vm.release(self.prepare)?;
                finish(vm, self, value, dependencies)
            }
            Err(error) => {
                // An invalid quantum can fail without discarding the VM's
                // continuation. No failed job may strand a read-only scope.
                let diagnostic = crate::endor_host::runtime_error(vm, error);
                self.cancel(vm)?;
                Err(diagnostic)
            }
        }
    }
}

pub fn start(
    vm: &mut Vm,
    desktop: ObjectRef,
    mounts: Vec<WeakObjectRef>,
    root: Arc<UiNode>,
    mut state: Draft,
    slice: usize,
) -> Result<Progress> {
    anyhow::ensure!(slice > 0, "layout slice must be positive");
    anyhow::ensure!(
        !vm.has_suspended_call() && !vm.object_store().is_read_only(),
        "cannot start layout while another read-only execution is active"
    );
    let frame = &mut state.frame;
    let runtime = &mut state.runtime;
    let cache = &mut state.cache;
    cache.hits = 0;
    cache.misses = 0;
    let live: HashSet<_> = mounts.iter().copied().collect();
    cache.entries.retain(|mount, _| live.contains(mount));
    // Preserve the existing native measurement cache and avoid VM/result
    // allocation for small trees without custom layout methods. Contexts already exist.
    let mut custom = false;
    for weak in &mounts {
        let mount = vm
            .upgrade(*weak)?
            .ok_or_else(|| anyhow::anyhow!("layout Mount expired"))?;
        let result = (|| {
            let block = vm.payload::<Mount>(mount, "Mount")?.block;
            let definition = vm
                .object_store()
                .type_of(block)?
                .ok_or(invalid("untyped layout target"))?;
            for name in ["measure", "arrange"] {
                let Some(method) = vm.object_store().lookup_method(definition, name)? else {
                    return Ok::<_, VmError>(true);
                };
                if !crate::native_visual::is_intrinsic(vm.object_store(), method, name, desktop)? {
                    return Ok(true);
                }
            }
            Ok(false)
        })();
        vm.release(mount)?;
        custom |= result?;
    }
    let mut nodes = Vec::new();
    let mut work = vec![root.clone()];
    while let Some(node) = work.pop() {
        custom |= node.has_flex() || node.allocates_children();
        work.extend(node.children().iter().rev().cloned());
        nodes.push(node);
    }
    // Flex allocation can change text wrapping and ancestor measurements even
    // when every protocol method is native. Use the same settling driver.
    if !custom && nodes.len() <= NATIVE_FAST_PATH_MAX_NODES {
        cache.clear();
        let mut native = runtime.frame(&root)?;
        // Only the structural pass consumes a published version.
        native.layout_version = frame.layout_version;
        runtime.version = frame.layout_version;
        *frame = native;
        state.reads.clear();
        return Ok(Progress::Complete(state));
    }
    anyhow::ensure!(
        nodes.len() == mounts.len() && nodes.len() == frame.mounts.len(),
        "layout structure does not match Mount contexts"
    );
    let indices: HashMap<_, _> = frame
        .mounts
        .iter()
        .enumerate()
        .map(|(index, mount)| (mount.id, index))
        .collect();
    let mut children = vec![Vec::new(); nodes.len()];
    for (index, mount) in frame.mounts.iter().enumerate() {
        if let Some(parent) = mount.parent {
            children[indices[&parent]].push(index);
        }
    }
    let mut constraints = vec![runtime.root_constraints(); nodes.len()];
    for index in 0..nodes.len() {
        for (child_index, child) in children[index].iter().enumerate() {
            constraints[*child] = layout::initial_child_constraints(
                &nodes[index],
                &nodes[*child],
                constraints[index],
                child_index,
            );
        }
    }
    let count = nodes.len();
    let plan = Arc::new(Mutex::new(Plan {
        sizes: vec![Size::default(); nodes.len()],
        preferred: vec![None; nodes.len()],
        bounds: vec![None; nodes.len()],
        nodes,
        children,
        runtime: runtime.clone(),
        custom: false,
        cache: cache.clone(),
        pending: (0..count).map(|_| None).collect(),
        constraints,
        allocations: vec![[None, None]; count],
        custom_arrange: vec![false; count],
        native_arrange: vec![false; count],
        custom_measure: vec![false; count],
        measured_limits: vec![[None, None]; count],
        previous: None,
        seen: BTreeSet::new(),
    }));
    let prepare = vm.allocate(ContextNativeFunction::new(
        "prepare source layout",
        Prepare {
            plan: plan.clone(),
            mounts,
            desktop,
        },
        vec![desktop],
    ))?;
    let work = Work {
        state,
        plan,
        prepare,
        desktop,
        slices: 1,
    };
    let progress = vm.start_read_only_call(
        prepare,
        &[],
        endor_core::ExecutionLimits {
            max_frames: 128,
            max_instructions: 1_000_000,
        },
        slice,
    );
    work.accept(vm, progress)
}

fn finish(
    vm: &mut Vm,
    mut work: Work,
    value: Value,
    reads: Vec<endor_core::ReadDependency>,
) -> Result<Progress> {
    vm.release_value(value)?;
    let mut plan = work
        .plan
        .lock()
        .map_err(|_| anyhow::anyhow!("layout plan poisoned"))?;
    plan.runtime.slices = work.slices;
    let frame = &mut work.state.frame;
    frame.commands.clear();
    for (index, mount) in frame.mounts.iter_mut().enumerate() {
        let [x, y, width, height] =
            plan.bounds[index].ok_or_else(|| anyhow::anyhow!("unarranged occurrence"))?;
        mount.x = x;
        mount.y = y;
        mount.width = width;
        mount.height = height;
        if let Some(command) =
            layout::paint_native(&plan.nodes[index], x, y, Size { width, height })
        {
            command.validate()?;
            frame.commands.push(command);
        }
    }
    work.state.runtime = plan.runtime.clone();
    work.state.cache = plan.cache.clone();
    if !plan.custom {
        work.state.reads.clear();
        return Ok(Progress::Complete(work.state));
    }
    let registry = match vm.get_attr(work.desktop, "types")? {
        Value::Object(object) => object,
        _ => anyhow::bail!("missing type registry"),
    };
    work.state.reads = reads
        .into_iter()
        .filter(|dependency| {
            let object = dependency.object();
            object != work.desktop
                && object != registry
                && vm
                    .object(object)
                    .is_ok_and(|object| !object.payload().as_any().is::<Mount>())
        })
        .collect();
    Ok(Progress::Complete(work.state))
}