//! Owner-managed asynchronous work. Only weak identities enter the scheduler;
//! callbacks and scopes are ordinary traced edges in the Aster object graph.
use std::{
    sync::{
        Arc, Mutex,
        atomic::{AtomicBool, Ordering},
    },
    time::{Duration, Instant},
};

use anyhow::{Result, bail};
use endor_core::{
    ContextNativeFunction, NativeCallable, NativeContext, NativeResult, ObjectPayload, ObjectRef,
    ObjectStore, Value, Vm, VmError, WeakObjectRef,
};

use crate::{block::Block, mount::Mount, ui_bridge::desktop_type, window::ManagedWindow};

const SCOPE: &str = "\0tessera.tasks";
const MAX_TASKS: usize = 1024;
const MAX_OWNER_TASKS: usize = 64;
const MAX_READY: usize = 64;

type Index = Arc<Mutex<Vec<WeakObjectRef>>>;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum State {
    Pending,
    Running,
    Completed,
    Cancelled,
    Failed,
}
impl State {
    fn name(self) -> &'static str {
        match self {
            Self::Pending => "pending",
            Self::Running => "running",
            Self::Completed => "completed",
            Self::Cancelled => "cancelled",
            Self::Failed => "failed",
        }
    }
}

#[derive(Debug, Clone)]
struct Task {
    owner: WeakObjectRef,
    key: String,
    due: Instant,
    callback: Option<ObjectRef>,
    state: State,
    ready: bool,
    text: Option<Arc<str>>,
    error: Option<Arc<str>>,
    lease: Option<Arc<crate::io_tasks::Lease>>,
    cooperative: bool,
    coroutine: Option<ObjectRef>,
    waiting: Option<ObjectRef>,
    result: Option<Value>,
}
impl ObjectPayload for Task {
    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
    fn kind_name(&self) -> &'static str {
        "tessera-task"
    }
    fn reference_assignment_only(&self) -> bool {
        true
    }
    fn poll_await(&self) -> Result<endor_core::AwaitPoll, VmError> {
        match self.state {
            State::Pending | State::Running => Ok(endor_core::AwaitPoll::Pending),
            State::Completed => Ok(endor_core::AwaitPoll::Ready(
                self.result.clone().unwrap_or_else(|| {
                    self.text
                        .as_ref()
                        .map(|text| Value::Text(text.to_string()))
                        .unwrap_or(Value::Nil)
                }),
            )),
            State::Cancelled => Err(VmError::AwaitCancelled),
            State::Failed => Err(VmError::AwaitFailed(
                self.error.as_deref().unwrap_or("task failed").into(),
            )),
        }
    }
    fn trace(&self, visit: &mut dyn FnMut(ObjectRef)) {
        if let Some(callback) = self.callback {
            visit(callback);
        }
        if let Some(coroutine) = self.coroutine {
            visit(coroutine);
        }
        if let Some(waiting) = self.waiting {
            visit(waiting);
        }
        if let Some(Value::Object(result)) = self.result {
            visit(result);
        }
    }
}
fn invalid(message: &'static str) -> VmError {
    VmError::InvalidArgument(message)
}
fn object(value: &Value) -> Result<ObjectRef, VmError> {
    match value {
        Value::Object(object) => Ok(*object),
        _ => Err(invalid("task owner and callback must be managed objects")),
    }
}
fn scope(store: &ObjectStore, owner: ObjectRef) -> Result<Option<ObjectRef>, VmError> {
    let payload = store.object(owner)?.payload();
    if let Some(value) = payload.as_any().downcast_ref::<Block>() {
        return Ok(value.tasks);
    }
    if let Some(value) = payload.as_any().downcast_ref::<Mount>() {
        return Ok(value.tasks);
    }
    if let Some(value) = payload.as_any().downcast_ref::<ManagedWindow>() {
        return Ok(value.tasks);
    }
    match store.get_attr(owner, SCOPE) {
        Ok(Value::Object(scope)) => {
            // Validate even the private host field before using it as an owner edge.
            store.dict_items(scope)?;
            Ok(Some(scope))
        }
        Err(VmError::AttributeNotFound(_)) => Ok(None),
        Ok(_) => Err(invalid("invalid task scope")),
        Err(error) => Err(error),
    }
}
fn install_scope(
    cx: &mut NativeContext<'_>,
    owner: ObjectRef,
    scope: ObjectRef,
) -> Result<(), VmError> {
    let payload = cx.store().object(owner)?.payload();
    if let Some(value) = payload.as_any().downcast_ref::<Block>() {
        let mut value = value.clone();
        value.tasks = Some(scope);
        return cx.replace_payload(owner, value);
    }
    if let Some(value) = payload.as_any().downcast_ref::<Mount>() {
        let mut value = value.clone();
        value.tasks = Some(scope);
        return cx.replace_payload(owner, value);
    }
    if let Some(value) = payload.as_any().downcast_ref::<ManagedWindow>() {
        let mut value = value.clone();
        value.tasks = Some(scope);
        return cx.replace_payload(owner, value);
    }
    cx.set_attr(owner, SCOPE, Value::Object(scope))
}
fn eligible(store: &ObjectStore, owner: ObjectRef, desktop: ObjectRef) -> Result<bool, VmError> {
    let payload = store.object(owner)?.payload();
    if let Some(mount) = payload.as_any().downcast_ref::<Mount>() {
        if !mount.mounted {
            return Ok(false);
        }
        if let Some(page) = mount.page {
            // No retained root is needed for this synchronous borrowed lookup.
            let windows = object(&store.get_attr(desktop, "windows")?)?;
            for (_, value) in store.dict_items(windows)? {
                for candidate in crate::window::pages(store, object(value)?)? {
                    if store.downgrade(*candidate)? == page {
                        return Ok(true);
                    }
                }
            }
            return Ok(false);
        }
        return Ok(false);
    }
    if let Some(window) = payload.as_any().downcast_ref::<ManagedWindow>() {
        return Ok(!window.closed);
    }
    if payload
        .as_any()
        .downcast_ref::<Block>()
        .is_some_and(|block| block.page)
    {
        let windows = object(&store.get_attr(desktop, "windows")?)?;
        for (_, value) in store.dict_items(windows)? {
            let window = object(value)?;
            if crate::window::pages(store, window)?.contains(&owner) {
                return Ok(true);
            }
        }
        return Ok(false);
    }
    Ok(true)
}
fn current(
    store: &ObjectStore,
    owner: ObjectRef,
    reference: ObjectRef,
    task: &Task,
) -> Result<bool, VmError> {
    Ok(match scope(store, owner)? {
        Some(scope) => {
            store.dict_get(scope, &Value::Text(task.key.clone()))? == Some(Value::Object(reference))
        }
        None => false,
    })
}

/// Called at the actual close/detach boundary, including close-then-reopen in
/// one language input. A later appearance never revives these identities.
pub fn cancel_owner(cx: &mut NativeContext<'_>, owner: ObjectRef) -> Result<(), VmError> {
    let Some(scope) = scope(cx.store(), owner)? else {
        return Ok(());
    };
    for (key, value) in cx.store().dict_items(scope)?.to_vec() {
        let reference = object(&value)?;
        let mut task = cx.store().payload::<Task>(reference, "Task")?.clone();
        if matches!(task.state, State::Pending | State::Running) {
            task.state = State::Cancelled;
            task.callback = None;
            task.lease = None;
            task.coroutine = None;
            task.waiting = None;
            task.text = None;
            task.result = None;
            cx.replace_payload(reference, task)?;
        }
        cx.dict_delete(scope, &key)?;
    }
    Ok(())
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Operation {
    Timer,
    Sleep,
    ReadText,
    FetchText,
    Spawn,
}
#[derive(Debug)]
struct After {
    operation: Operation,
    provider: crate::io_tasks::Provider,
    signature: endor_core::CallSignature,
    index: Index,
    runnable: Arc<AtomicBool>,
    quiescing: Arc<AtomicBool>,
    desktop: ObjectRef,
}
impl NativeCallable for After {
    fn signature(&self) -> Option<endor_core::CallSignature> {
        Some(self.signature.clone())
    }
    fn arity(&self) -> Option<endor_core::CallArity> {
        Some(self.signature.arity)
    }

    fn invoke(&self, cx: &mut NativeContext<'_>, args: &[Value]) -> Result<NativeResult, VmError> {
        if self.quiescing.load(Ordering::Acquire) {
            return Err(invalid("runtime is exiting; new tasks are forbidden"));
        }
        if cx.store().is_read_only() {
            return Err(invalid(
                "asynchronous work is forbidden during read-only rendering",
            ));
        }
        let zero = Value::Number(0_i64.into());
        let (owner, key, delay, callback) = match args {
            [owner, key, callback] if self.operation == Operation::Spawn => {
                (owner, key, &zero, Some(callback))
            }
            [owner, key, delay] if self.operation == Operation::Sleep => (owner, key, delay, None),
            [owner, key, delay, callback] => (owner, key, delay, Some(callback)),
            [owner, key, path]
                if matches!(self.operation, Operation::ReadText | Operation::FetchText) =>
            {
                (owner, key, path, None)
            }
            _ => {
                return Err(invalid(
                    "asynchronous request requires owner, key, delay/path and optional I/O callback",
                ));
            }
        };
        let owner = object(owner)?;
        let callback = callback.map(object).transpose()?;
        if let Some(callback) = callback {
            if !cx.store().object(callback)?.payload().is_function() {
                return Err(invalid("task callback must be a Function"));
            }
            if !cx
                .store()
                .call_arity(callback)?
                .ok_or(invalid("task callback requires positional metadata"))?
                .accepts(0)
            {
                return Err(invalid("task callback requires zero unbound parameters"));
            }
        }
        let Value::Text(key) = key else {
            return Err(invalid("task key must be text"));
        };
        if key.is_empty() || key.len() > 128 || key.chars().any(char::is_control) {
            return Err(invalid("invalid task key"));
        }
        let (milliseconds, request) =
            if matches!(self.operation, Operation::ReadText | Operation::FetchText) {
                let Value::Text(path) = delay else {
                    return Err(invalid("I/O input must be a String"));
                };
                let request = match self.operation {
                    Operation::ReadText => {
                        if path.is_empty() || path.len() > 4096 || path.contains('\0') {
                            return Err(invalid("invalid text path"));
                        }
                        crate::io_tasks::Request::ReadText(path.clone())
                    }
                    Operation::FetchText => {
                        if !crate::io_tasks::valid_url(path) {
                            return Err(invalid(
                                "expected HTTP(S) URL without credentials or control characters",
                            ));
                        }
                        crate::io_tasks::Request::FetchText(path.clone())
                    }
                    Operation::Timer | Operation::Sleep | Operation::Spawn => unreachable!(),
                };
                (0, Some(request))
            } else {
                let Value::Number(delay) = delay else {
                    return Err(invalid(
                        "task delay must be an integer number of milliseconds",
                    ));
                };
                let milliseconds = delay
                    .as_i64()
                    .filter(|delay| (0..=86_400_000).contains(delay))
                    .ok_or(invalid(
                        "task delay must be 0..86400000 integer milliseconds",
                    ))?;
                (milliseconds, None)
            };
        if !eligible(cx.store(), owner, self.desktop)? {
            return Err(invalid("task owner is closed or detached"));
        }
        let existing_scope = scope(cx.store(), owner)?;
        let previous = existing_scope
            .map(|scope| cx.store().dict_get(scope, &Value::Text(key.clone())))
            .transpose()?
            .flatten()
            .map(|value| object(&value))
            .transpose()?;
        let mut index = self
            .index
            .lock()
            .map_err(|_| invalid("task index poisoned"))?;
        let mut live = Vec::new();
        for weak in index.iter().copied() {
            if let Some(task) = cx.upgrade(weak)?
                && matches!(
                    cx.store().payload::<Task>(task, "Task")?.state,
                    State::Pending | State::Running
                )
            {
                live.push(weak);
            }
        }
        if live.len() >= MAX_TASKS && previous.is_none() {
            return Err(VmError::ExecutionLimit("1024 pending tasks per runtime"));
        }
        if let Some(scope) = existing_scope
            && cx.store().dict_items(scope)?.len() >= MAX_OWNER_TASKS
            && previous.is_none()
        {
            return Err(VmError::ExecutionLimit("64 pending tasks per owner"));
        }
        let task = cx.allocate(Task {
            owner: cx.store().downgrade(owner)?,
            key: key.clone(),
            due: Instant::now()
                .checked_add(Duration::from_millis(milliseconds as u64))
                .ok_or(invalid("task deadline overflow"))?,
            callback,
            state: State::Pending,
            ready: matches!(
                self.operation,
                Operation::Timer | Operation::Sleep | Operation::Spawn
            ),
            text: None,
            error: None,
            lease: None,
            cooperative: self.operation == Operation::Spawn,
            coroutine: None,
            waiting: None,
            result: None,
        })?;
        cx.attach_type(task, desktop_type(cx.store(), "Task", self.desktop)?)?;
        if let Some(request) = request {
            let replacing = previous
                .map(|previous| {
                    cx.store().payload::<Task>(previous, "Task").map(|task| {
                        matches!(task.state, State::Pending | State::Running)
                            && task.lease.is_some()
                    })
                })
                .transpose()?
                .unwrap_or(false);
            let lease = self
                .provider
                .submit(cx.store().downgrade(task)?, request, replacing)?;
            let mut value = cx.store().payload::<Task>(task, "Task")?.clone();
            value.lease = Some(lease);
            cx.replace_payload(task, value)?;
        }

        let task_scope = if let Some(scope) = existing_scope {
            scope
        } else {
            let scope = cx.new_dict()?;
            install_scope(cx, owner, scope)?;
            scope
        };
        // Publish the new identity before invalidating the old result. Nothing
        // external is dispatched until the worker accepts this VM generation.
        let previous_weak = previous.map(|old| cx.store().downgrade(old)).transpose()?;
        if let Some(weak) = previous_weak {
            cx.upgrade(weak)?;
        }
        cx.dict_set(task_scope, Value::Text(key.clone()), Value::Object(task))?;
        if let Some(previous) = previous {
            // The replacement above can release the old Task's last owner edge.
            // Preserve it through the replacement even without a saved token.
            if cx.store().object(previous).is_ok() {
                let mut old = cx.store().payload::<Task>(previous, "Task")?.clone();
                if matches!(old.state, State::Pending | State::Running) {
                    old.state = State::Cancelled;
                    old.callback = None;
                    old.lease = None;
                    old.coroutine = None;
                    old.waiting = None;
                    old.text = None;
                    old.result = None;
                    cx.replace_payload(previous, old)?;
                }
            }
        }
        let weak = cx.store().downgrade(task)?;
        live.retain(|old| Some(*old) != previous_weak);
        live.push(weak);
        *index = live;
        if self.operation == Operation::Spawn {
            self.runnable.store(true, Ordering::Release);
        }
        Ok(NativeResult::Return(Value::Object(task)))
    }
}

#[derive(Debug, Clone, Copy)]
enum Query {
    Cancel,
    Status,
    Active,
    Result,
    Error,
}
#[derive(Debug)]
struct Method {
    signature: endor_core::CallSignature,
    query: Query,
    desktop: ObjectRef,
}
impl NativeCallable for Method {
    fn signature(&self) -> Option<endor_core::CallSignature> {
        Some(self.signature.clone())
    }
    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 [value] = args else {
            return Err(invalid("Task methods accept no arguments"));
        };
        let reference = object(value)?;
        let mut task = cx.store().payload::<Task>(reference, "Task")?.clone();
        let pending = matches!(task.state, State::Pending | State::Running);
        let owner = if pending {
            cx.upgrade(task.owner)?
        } else {
            None
        };
        let active = if let Some(owner) = owner {
            eligible(cx.store(), owner, self.desktop)?
                && current(cx.store(), owner, reference, &task)?
        } else {
            false
        };
        let value = match self.query {
            Query::Status => Value::Text(
                if pending && !active {
                    State::Cancelled.name()
                } else {
                    task.state.name()
                }
                .into(),
            ),
            Query::Active => Value::Bool(active),
            Query::Result => {
                if !matches!(task.state, State::Running | State::Completed)
                    || task.error.is_some()
                    || (task.cooperative && task.state != State::Completed)
                {
                    return Err(invalid(
                        "task result is not available; inspect status and error",
                    ));
                }
                task.result.clone().unwrap_or_else(|| {
                    task.text
                        .as_ref()
                        .map(|text| Value::Text(text.to_string()))
                        .unwrap_or(Value::Nil)
                })
            }
            Query::Error => Value::Text(
                if task.state == State::Failed {
                    task.error.as_deref().unwrap_or_default()
                } else {
                    ""
                }
                .into(),
            ),
            Query::Cancel => {
                if cx.store().is_read_only() {
                    return Err(invalid(
                        "task cancellation is forbidden during read-only rendering",
                    ));
                }
                if pending {
                    task.state = State::Cancelled;
                    task.callback = None;
                    task.lease = None;
                    task.coroutine = None;
                    task.waiting = None;
                    task.text = None;
                    task.result = None;
                    cx.replace_payload(reference, task.clone())?;
                    if let Some(owner) = owner
                        && current(cx.store(), owner, reference, &task)?
                        && let Some(scope) = scope(cx.store(), owner)?
                    {
                        cx.dict_delete(scope, &Value::Text(task.key))?;
                    }
                }
                Value::Bool(pending && active)
            }
        };
        Ok(NativeResult::Return(value))
    }
}

pub struct Scheduler {
    index: Index,
    runnable: Arc<AtomicBool>,
    quiescing: Arc<AtomicBool>,
    desktop: ObjectRef,
    provider: crate::io_tasks::Provider,
}
impl Scheduler {
    pub fn quiesce(&self, vm: &mut Vm) -> Result<usize> {
        self.quiescing.store(true, Ordering::Release);
        self.cancel_all(vm)
    }
    pub fn has_cooperative_work(&self) -> bool {
        self.runnable.load(Ordering::Acquire)
    }
    /// Mutation boundaries collect unreachable owner/callback cycles before
    /// this weak lookup. Idle timer polling does not rescan the entire heap.
    pub fn poll(&self, vm: &mut Vm, limit: usize) -> Result<usize> {
        anyhow::ensure!((1..=MAX_READY).contains(&limit), "invalid task poll limit");
        // Provider threads only fill the mailbox. Only this VM worker updates
        // managed payloads, and delivery below rechecks owner/key generation.
        for (weak, completion) in self.provider.take_ready()? {
            if let Some(reference) = vm.upgrade(weak)? {
                let result = (|| -> Result<()> {
                    let mut task = vm.payload::<Task>(reference, "Task")?.clone();
                    if task.state != State::Pending || task.lease.is_none() {
                        return Ok(());
                    }
                    let owner = vm.upgrade(task.owner)?;
                    let active = (|| -> Result<bool> {
                        Ok(if let Some(owner) = owner {
                            eligible(vm.object_store(), owner, self.desktop)?
                                && current(vm.object_store(), owner, reference, &task)?
                        } else {
                            false
                        })
                    })();
                    if let Some(owner) = owner {
                        vm.release(owner)?;
                    }
                    if !active? {
                        return Ok(());
                    }
                    task.ready = true;
                    match completion {
                        Ok(text) => task.text = Some(text.into()),
                        Err(error) => task.error = Some(error.into()),
                    }
                    vm.replace_payload(reference, task)?;
                    Ok(())
                })();
                vm.release(reference)?;
                result?;
            }
        }
        let snapshot = self
            .index
            .lock()
            .map_err(|_| anyhow::anyhow!("task index poisoned"))?
            .clone();
        let now = Instant::now();
        let mut work = 0;
        let mut advanced = Vec::new();
        for weak in snapshot {
            if work >= limit {
                break;
            }
            let Some(reference) = vm.upgrade(weak)? else {
                continue;
            };
            let result = (|| -> Result<()> {
                let mut task = vm.payload::<Task>(reference, "Task")?.clone();
                if !matches!(task.state, State::Pending | State::Running) {
                    return Ok(());
                }
                let owner = vm.upgrade(task.owner)?;
                let result = (|| -> Result<()> {
                    let active = if let Some(owner) = owner {
                        eligible(vm.object_store(), owner, self.desktop)?
                            && current(vm.object_store(), owner, reference, &task)?
                    } else {
                        false
                    };
                    if !active {
                        work += 1;
                        return finish(vm, reference, State::Cancelled);
                    }
                    if let Some(waiting) = task.waiting {
                        match vm.object(waiting)?.payload().poll_await() {
                            Ok(endor_core::AwaitPoll::Pending) => return Ok(()),
                            Err(VmError::AwaitCancelled) => {
                                work += 1;
                                return finish(vm, reference, State::Cancelled);
                            }
                            _ => {}
                        }
                    }
                    if task.due > now || !task.ready {
                        return Ok(());
                    }
                    work += 1;
                    advanced.push(weak);
                    if task.cooperative {
                        return advance_coroutine(vm, reference);
                    }
                    if task.error.is_some() {
                        return finish(vm, reference, State::Failed);
                    }
                    let Some(callback) = task.callback else {
                        return finish(vm, reference, State::Completed);
                    };
                    // The provider is finished. Keep the result in the Task,
                    // but release its I/O slot before running source callbacks.
                    task.lease = None;
                    task.coroutine = None;
                    task.waiting = None;
                    let mut running = task.clone();
                    running.state = State::Running;
                    vm.replace_payload(reference, running)?;
                    let result = vm
                        .call_with_limits(
                            callback,
                            &[],
                            endor_core::ExecutionLimits {
                                max_frames: 128,
                                max_instructions: 100_000,
                            },
                        )
                        .map_err(|error| crate::endor_host::runtime_error(vm, error));
                    match result {
                        Ok(value) => {
                            let unit = matches!(value, Value::Nil);
                            vm.release_value(value)?;
                            if !unit {
                                fail_callback(
                                    vm,
                                    reference,
                                    "asynchronous callback must return Unit",
                                )?;
                                bail!("asynchronous callback must return Unit");
                            }
                            finish(vm, reference, State::Completed)
                        }
                        Err(error) => {
                            fail_callback(vm, reference, &format!("{error:#}"))?;
                            Err(error)
                        }
                    }
                })();
                if let Some(owner) = owner {
                    vm.release(owner)?;
                }
                result
            })();
            vm.release(reference)?;
            result?;
        }
        // Round-robin resumable work so a long call cannot monopolize the
        // one-callback idle budget and starve other runnable Tasks.
        if !advanced.is_empty() {
            let mut index = self
                .index
                .lock()
                .map_err(|_| anyhow::anyhow!("task index poisoned"))?;
            index.retain(|weak| !advanced.contains(weak));
            index.extend(advanced);
        }
        self.prune(vm)?;
        Ok(work)
    }
    pub fn cancel_all(&self, vm: &mut Vm) -> Result<usize> {
        let snapshot = self
            .index
            .lock()
            .map_err(|_| anyhow::anyhow!("task index poisoned"))?
            .clone();
        for weak in snapshot {
            if let Some(task) = vm.upgrade(weak)? {
                let result = finish(vm, task, State::Cancelled);
                vm.release(task)?;
                result?;
            }
        }
        let mut index = self
            .index
            .lock()
            .map_err(|_| anyhow::anyhow!("task index poisoned"))?;
        let count = index.len();
        index.clear();
        self.runnable.store(false, Ordering::Release);
        Ok(count)
    }
    fn prune(&self, vm: &mut Vm) -> Result<()> {
        let mut index = self
            .index
            .lock()
            .map_err(|_| anyhow::anyhow!("task index poisoned"))?;
        let mut live = Vec::new();
        let mut runnable = false;
        for weak in index.iter().copied() {
            if let Some(task) = vm.upgrade(weak)? {
                let pending = vm.payload::<Task>(task, "Task").map(|task| {
                    let pending = matches!(task.state, State::Pending | State::Running);
                    runnable |= pending && task.cooperative && task.waiting.is_none();
                    pending
                });
                if pending.as_ref().is_ok_and(|pending| *pending) {
                    let value = vm.payload::<Task>(task, "Task")?;
                    if value.cooperative
                        && let Some(waiting) = value.waiting
                    {
                        runnable |= !matches!(
                            vm.object(waiting)?.payload().poll_await(),
                            Ok(endor_core::AwaitPoll::Pending)
                        );
                    }
                }
                vm.release(task)?;
                if pending? {
                    live.push(weak);
                }
            }
        }
        *index = live;
        self.runnable.store(runnable, Ordering::Release);
        Ok(())
    }
}
fn cyclic_wait(vm: &Vm, source: ObjectRef, mut target: ObjectRef) -> Result<bool> {
    let mut seen = std::collections::HashSet::new();
    loop {
        if target == source || !seen.insert(target) {
            return Ok(true);
        }
        let Some(task) = vm.object(target)?.payload().as_any().downcast_ref::<Task>() else {
            return Ok(false);
        };
        let Some(waiting) = task.waiting else {
            return Ok(false);
        };
        target = waiting;
    }
}

fn advance_coroutine(vm: &mut Vm, reference: ObjectRef) -> Result<()> {
    let mut task = vm.payload::<Task>(reference, "Task")?.clone();
    let coroutine = if let Some(coroutine) = task.coroutine {
        coroutine
    } else {
        let callback = task
            .callback
            .ok_or_else(|| anyhow::anyhow!("spawn callback missing"))?;
        let prepared = vm
            .new_coroutine(
                callback,
                &[],
                endor_core::ExecutionLimits {
                    max_frames: 128,
                    max_instructions: 100_000,
                },
            )
            .map_err(|error| crate::endor_host::runtime_error(vm, error));
        let coroutine = match prepared {
            Ok(coroutine) => coroutine,
            Err(error) => {
                fail_callback(vm, reference, &format!("{error:#}"))?;
                return Err(error);
            }
        };
        // Native preparation can itself cancel or replace this Task.
        let current = vm.payload::<Task>(reference, "Task")?.state;
        if !matches!(current, State::Pending | State::Running) {
            vm.release(coroutine)?;
            return Ok(());
        }
        task.state = State::Running;
        task.callback = None;
        task.coroutine = Some(coroutine);
        let installed = vm.replace_payload(reference, task);
        vm.release(coroutine)?;
        installed?;
        coroutine
    };
    let result = vm
        .resume_coroutine(coroutine, 256)
        .map_err(|error| crate::endor_host::runtime_error(vm, error));
    // Cancellation/replacement from the running function must win. A completed
    // value still owns a root and must be released even when delivery is stale.
    let active = matches!(
        vm.payload::<Task>(reference, "Task")?.state,
        State::Pending | State::Running
    );
    match result {
        Ok(endor_core::CoroutineProgress::Waiting { awaitable, .. }) => {
            if !active {
                return Ok(());
            }
            if cyclic_wait(vm, reference, awaitable)? {
                fail_callback(vm, reference, "cyclic task await")?;
                bail!("cyclic task await");
            }
            let mut task = vm.payload::<Task>(reference, "Task")?.clone();
            task.waiting = Some(awaitable);
            vm.replace_payload(reference, task)?;
            Ok(())
        }
        Ok(endor_core::CoroutineProgress::Pending { .. }) => {
            if active {
                let mut task = vm.payload::<Task>(reference, "Task")?.clone();
                task.waiting = None;
                vm.replace_payload(reference, task)?;
            }
            Ok(())
        }
        Ok(endor_core::CoroutineProgress::Complete { value, .. }) => {
            if !active {
                vm.release_value(value)?;
                return Ok(());
            }
            let mut task = vm.payload::<Task>(reference, "Task")?.clone();
            task.result = Some(value.clone());
            task.waiting = None;
            let stored = vm.replace_payload(reference, task);
            vm.release_value(value)?;
            stored?;
            finish(vm, reference, State::Completed)
        }
        Err(error) => {
            if active {
                fail_callback(vm, reference, &format!("{error:#}"))?;
            }
            Err(error)
        }
    }
}

fn fail_callback(vm: &mut Vm, reference: ObjectRef, message: &str) -> Result<()> {
    let mut task = vm.payload::<Task>(reference, "Task")?.clone();
    task.error = Some(message.chars().take(1024).collect::<String>().into());
    vm.replace_payload(reference, task)?;
    finish(vm, reference, State::Failed)
}

fn finish(vm: &mut Vm, reference: ObjectRef, state: State) -> Result<()> {
    let mut task = vm.payload::<Task>(reference, "Task")?.clone();
    if !matches!(task.state, State::Pending | State::Running) {
        return Ok(());
    }
    task.state = state;
    task.callback = None;
    task.lease = None;
    task.coroutine = None;
    task.waiting = None;
    if state == State::Cancelled {
        task.text = None;
        task.result = None;
    }
    vm.replace_payload(reference, task.clone())?;
    if let Some(owner) = vm.upgrade(task.owner)? {
        let result = (|| -> Result<()> {
            if current(vm.object_store(), owner, reference, &task)?
                && let Some(scope) = scope(vm.object_store(), owner)?
            {
                vm.dict_delete(scope, &Value::Text(task.key))?;
            }
            Ok(())
        })();
        vm.release(owner)?;
        result?;
    }
    Ok(())
}

pub fn install(
    vm: &mut Vm,
    roots: &mut Vec<ObjectRef>,
    ui: ObjectRef,
    desktop: ObjectRef,
) -> Result<Scheduler> {
    let index = Arc::new(Mutex::new(Vec::new()));
    let runnable = Arc::new(AtomicBool::new(false));
    let quiescing = Arc::new(AtomicBool::new(false));
    let definition = desktop_type(vm.object_store(), "Task", desktop)?;
    let provider = crate::io_tasks::Provider::default();
    for (name, operation) in [
        ("after", Operation::Timer),
        ("sleep", Operation::Sleep),
        ("readText", Operation::ReadText),
        ("fetchText", Operation::FetchText),
        ("spawn", Operation::Spawn),
    ] {
        let external = matches!(operation, Operation::ReadText | Operation::FetchText);
        let function = vm.allocate(ContextNativeFunction::new(
            name,
            After {
                operation,
                provider: provider.clone(),
                signature: {
                    let mut signature = endor_core::CallSignature::exact(
                        vec![
                            endor_core::AnnotationType::Any,
                            endor_core::AnnotationType::String,
                            if external {
                                endor_core::AnnotationType::String
                            } else {
                                endor_core::AnnotationType::Number
                            },
                            endor_core::AnnotationType::Function,
                        ],
                        endor_core::AnnotationType::Nominal(definition),
                        if external {
                            endor_core::CallEffect::External
                        } else {
                            endor_core::CallEffect::Mutating
                        },
                    );
                    if external {
                        signature.arity = endor_core::CallArity::range(3, 4);
                    }
                    if operation == Operation::Spawn {
                        signature.parameters.remove(2);
                        signature.arity = endor_core::CallArity::exact(3);
                    }
                    if operation == Operation::Sleep {
                        signature.parameters.pop();
                        signature.arity = endor_core::CallArity::exact(3);
                    }
                    signature
                },
                index: index.clone(),
                runnable: runnable.clone(),
                quiescing: quiescing.clone(),
                desktop,
            },
            vec![desktop],
        ))?;
        roots.push(function);
        vm.set_attr(ui, name, Value::Object(function))?;
    }
    for (name, query) in [
        ("cancel", Query::Cancel),
        ("status", Query::Status),
        ("active", Query::Active),
        ("result", Query::Result),
        ("error", Query::Error),
    ] {
        let method = vm.allocate(ContextNativeFunction::new(
            name,
            Method {
                query,
                desktop,
                signature: endor_core::CallSignature::exact(
                    vec![endor_core::AnnotationType::Nominal(definition)],
                    match query {
                        Query::Status | Query::Error => endor_core::AnnotationType::String,
                        Query::Result => endor_core::AnnotationType::Any,
                        _ => endor_core::AnnotationType::Bool,
                    },
                    if matches!(query, Query::Cancel) {
                        endor_core::CallEffect::Mutating
                    } else {
                        endor_core::CallEffect::ReadOnly
                    },
                ),
            },
            vec![desktop],
        ))?;
        roots.push(method);
        vm.define_method(definition, name, method)?;
    }
    Ok(Scheduler {
        index,
        runnable,
        quiescing,
        desktop,
        provider,
    })
}