//! 唯一串行执行 VM 的 API worker。保留活跃运行时，另建 PUT 候选，并在候选完整成功后切换发布。
//! 输入、视口刷新与候选布局可以跨 worker 轮次挂起；中途只公布完整文档，不能让新输入进入未结束的只读调用。
//! 事件身份同时校验窗口、session、layout_version 和 Mount 能力。失败、替换与关闭分别处理，避免旧回调或旧命令泄漏到新 VM。
//!
//! Serial VM ownership, incremental requests, publication and shutdown.
use super::*;

type HttpResponse = Response<std::io::Cursor<Vec<u8>>>;
type InputOutcome =
    std::thread::Result<Result<(endor_host::ReplResult, Vec<crate::window::Command>)>>;

type CandidateOutcome = std::thread::Result<
    Result<(
        endor_host::ReplSession,
        crate::document::RenderDocument,
        Vec<crate::window::Command>,
    )>,
>;

/// 已执行业务部分但仍在等待展示刷新的输入元数据，避免恢复时重放回调。
pub(super) struct PendingInput {
    epoch: u64,
    id: u64,
    event: bool,
    result: endor_host::ReplResult,
}

/// 挂起 HTTP 请求的阶段：等旧刷新、继续输入或继续隔离候选。
pub(super) enum PendingRequest {
    BeforeInput,
    Input(Box<PendingInput>),
    Candidate(Box<endor_host::ReplSession>),
}

/// 一次 worker 推进的结果：继续保留请求或交付最终响应。
pub(super) enum RequestProgress {
    Pending(PendingRequest),
    Complete(HttpResponse),
}

/// 只提取仍属于当前 session 的窗口尺寸，避免旧原生窗口给新候选提供错误约束。
fn current_viewports(
    current: &NamespaceState,
) -> std::collections::BTreeMap<u64, crate::layout::Size> {
    let mut viewports: std::collections::BTreeMap<_, _> = current
        .window_viewports
        .iter()
        .filter(|(id, _)| {
            **id == crate::window::MAIN_WINDOW
                || (current.runtime_session != 0
                    && current
                        .document
                        .windows
                        .get(id)
                        .is_some_and(|view| view.frame.session == current.runtime_session))
        })
        .map(|(id, size)| (*id, *size))
        .collect();
    if let Some(viewport) = current.viewport {
        viewports.insert(crate::window::MAIN_WINDOW, viewport);
    }
    viewports
}

#[derive(Default)]
/// 唯一活跃 VM 的串行拥有者，另行管理候选和下一 session 身份。
pub(super) struct Worker {
    pub(super) repl: Option<(u64, endor_host::ReplSession)>,
    pub(super) next_session: u64,
}
impl Worker {
    /// 撤销活跃运行时并有界清理窗口、Mount 和任务；故障域不再执行新的业务回调。
    pub(super) fn shutdown_runtime(&mut self) -> ShutdownOutcome {
        let Some((_, mut runtime)) = self.repl.take() else {
            return ShutdownOutcome {
                summary: "Tessera shutdown: no live runtime".into(),
                error: None,
            };
        };
        let callbacks = !runtime.is_faulted();
        let result = catch_unwind(AssertUnwindSafe(|| runtime.shutdown()));
        match result {
            Ok(Ok((windows, mounts, tasks))) => ShutdownOutcome {
                summary: format!(
                    "Tessera shutdown: {windows} windows, {mounts} mounts, {tasks} task entries; callbacks={callbacks}"
                ),
                error: None,
            },
            result => {
                let error = match result {
                    Ok(Err(error)) => format!("shutdown cleanup failed: {error:#}"),
                    _ => "shutdown cleanup failed after an internal panic".into(),
                };
                let error: String = error.chars().take(4096).collect();
                ShutdownOutcome {
                    summary: error.clone(),
                    error: Some(error),
                }
            }
        }
    }

    /// 处理当前无法继续正常交互的窗口关闭，撤销原生资源与相应运行时记录。
    pub(super) fn close_inactive_window(
        &mut self,
        input: &crate::events::Input,
        epoch: u64,
        namespace: &Arc<RwLock<NamespaceState>>,
        notify: &Notify,
    ) -> Response<std::io::Cursor<Vec<u8>>> {
        if input.window == crate::window::MAIN_WINDOW {
            return error_response(
                StatusCode(409),
                "close the primary shell with its OS window control",
            );
        }
        let owns_runtime = self.repl.as_ref().is_some_and(|(owner, runtime)| {
            *owner == epoch && runtime.id == input.session && input.session != 0
        });
        let mut discarded = false;
        if owns_runtime {
            let result = catch_unwind(AssertUnwindSafe(|| {
                self.repl
                    .as_mut()
                    .expect("owned runtime")
                    .1
                    .retire_window(input.window)
            }));
            if !matches!(result, Ok(Ok(()))) {
                // Never keep a half-cleaned VM live or borrow another generation.
                self.repl = None;
                discarded = true;
            }
        }
        let Ok(mut current) = namespace.write() else {
            return error_response(StatusCode(500), "namespace lock poisoned");
        };
        if current.submission != epoch {
            return superseded(epoch);
        }
        let Some(view) = current.document.windows.get(&input.window) else {
            return error_response(StatusCode(409), "window is no longer active");
        };
        if view.frame.session != input.session {
            return error_response(
                StatusCode(409),
                "window belongs to another runtime generation",
            );
        }
        let mut document = current.document.clone();
        document.windows.remove(&input.window);
        if discarded {
            document.frame.session = 0;
            document.frame.focus = None;
            for view in document.windows.values_mut() {
                view.frame.session = 0;
                view.frame.focus = None;
            }
            current.runtime_session = 0;
            current.window_commands.clear();
        }
        match current.commit(epoch, document) {
            Ok(Some(_)) => {}
            Ok(None) => return superseded(epoch),
            Err(error) => return error_response(StatusCode(500), error.to_string()),
        }
        current.window_viewports.remove(&input.window);
        drop(current);
        notify();
        json(
            StatusCode(200),
            serde_json::json!({"ok":true,"execution":"window-retire","window":input.window,"callbacks":false,"runtime_discarded":discarded}),
        )
    }

    /// 在没有前台输入的 worker 轮次推进尺寸和任务引起的展示刷新。
    pub(super) fn refresh_viewport(
        &mut self,
        namespace: &Arc<RwLock<NamespaceState>>,
        notify: &Notify,
    ) {
        self.refresh_viewport_mode(namespace, notify, false);
    }

    /// 在输入之间推进视口/任务刷新；提交序号或 session 已变时拒绝旧结果发布。
    pub(super) fn refresh_viewport_mode(
        &mut self,
        namespace: &Arc<RwLock<NamespaceState>>,
        notify: &Notify,
        before_input: bool,
    ) {
        // 在读锁内取得一个一致的提交身份和尺寸快照；耗时 VM 工作不占用共享状态写锁。
        let (epoch, viewports) = match namespace.read() {
            Ok(current) => {
                let viewports = current_viewports(&current);
                (current.submission, viewports)
            }
            Err(_) => return,
        };
        let Some((owner, runtime)) = &mut self.repl else {
            return;
        };
        if *owner != epoch {
            if let Ok(current) = namespace.read() {
                self.preserve_live_owner(&current);
            }
            return;
        }
        let id = runtime.id;
        let result = catch_unwind(AssertUnwindSafe(|| -> Result<_> {
            if runtime.is_faulted() {
                runtime.poll_tasks()?;
                return Ok((None, Vec::new()));
            }
            let resized = runtime.resize_windows_incremental(&viewports, false)?;
            if runtime.has_pending_refresh() {
                return Ok((None, Vec::new()));
            }
            // An accepted foreground input waits only for the existing
            // refresh; do not keep placing new callbacks ahead of it.
            let completed = if before_input {
                None
            } else {
                runtime.poll_tasks_incremental(&viewports)?
            };
            if runtime.has_pending_refresh() {
                // A complete earlier resize may publish; task effects and their
                // native commands wait until the task's refresh has settled.
                return Ok((resized, Vec::new()));
            }
            Ok((completed.or(resized), runtime.take_window_commands()?))
        }));
        // 计算结束后重新核对归属：期间若发生 PUT/DELETE，不能把旧域草稿发布出去。
        let mut current = match namespace.write() {
            Ok(current) => current,
            Err(_) => return,
        };
        if current.submission != epoch || current.runtime_session != id {
            return;
        }
        let (document, commands) = match result {
            Ok(Ok((document, commands))) => {
                if let Some(document) = document {
                    (document, Some(commands))
                } else {
                    let projected = !commands.is_empty();
                    current.publish_window_commands(id, commands);
                    drop(current);
                    if projected {
                        notify();
                    }
                    return;
                }
            }
            failure => {
                let mut document = current.document.clone();
                document.frame.fault = Some(match failure {
                    Ok(Err(error)) => format!("Runtime background update failed: {error:#}")
                        .chars()
                        .take(1024)
                        .collect(),
                    _ => {
                        self.repl = None;
                        current.runtime_session = 0;
                        current.window_commands.clear();
                        "Runtime discarded after an internal background failure".into()
                    }
                });
                for view in document.windows.values_mut() {
                    view.frame.fault = document.frame.fault.clone();
                }
                (document, None)
            }
        };
        let changed = matches!(current.commit(epoch, document), Ok(Some(_)));
        if changed && let Some(commands) = commands {
            current.publish_window_commands(id, commands);
        }
        drop(current);
        if changed {
            notify();
        }
    }

    /// 检查请求种类和归属，必要时先等已有刷新结束，再建立输入或候选任务。
    pub(super) fn begin_request(
        &mut self,
        job: &BuildRequest,
        namespace: &Arc<RwLock<NamespaceState>>,
        notify: &Notify,
    ) -> RequestProgress {
        if job.kind == RequestKind::Code {
            return self.begin_program_request(job, namespace, notify);
        }
        if job.kind == RequestKind::Expression {
            return RequestProgress::Complete(build(job, namespace, notify));
        }
        // Advance one existing refresh quantum, then park the HTTP request.
        // No input statement executes inside the read-only scope; shutdown and
        // newer viewport targets can be observed at every worker turn.
        self.refresh_viewport_mode(namespace, notify, true);
        let epoch = match namespace.read() {
            Ok(current) => current.submission,
            Err(_) => {
                return RequestProgress::Complete(error_response(
                    StatusCode(500),
                    "namespace lock is poisoned",
                ));
            }
        };
        if epoch != job.submission {
            return RequestProgress::Complete(superseded(job.submission));
        }
        if job.kind == RequestKind::ResetRepl {
            self.repl = None;
            if let Ok(mut current) = namespace.write() {
                current.document.frame.session = 0;
                current.document.frame.focus = None;
                for document in current.document.windows.values_mut() {
                    document.frame.session = 0;
                    document.frame.focus = None;
                }
                current.runtime_session = 0;
                current.window_commands.clear();
            }
            notify();
            return RequestProgress::Complete(json(
                StatusCode(200),
                serde_json::json!({"ok":true,"execution":"repl-reset","persistent":true,"session_reset":true,"message":"Console memory cleared; displayed page preserved"}),
            ));
        }
        if self
            .repl
            .as_ref()
            .is_some_and(|(_, runtime)| runtime.has_pending_refresh())
        {
            return RequestProgress::Pending(PendingRequest::BeforeInput);
        }
        let event = if job.kind == RequestKind::Event {
            let mut input: crate::events::Input = if let Some(input) = &job.native {
                input.clone()
            } else {
                match serde_json::from_str(&job.source) {
                    Ok(input) => input,
                    Err(error) => {
                        return RequestProgress::Complete(error_response(
                            StatusCode(400),
                            error.to_string(),
                        ));
                    }
                }
            };
            if !input.kind.starts_with("developer-open:")
                && !matches!(
                    input.kind.as_str(),
                    "click"
                        | "pointermove"
                        | "pointerdown"
                        | "pointerup"
                        | "pointerleave"
                        | "pointercancel"
                        | "focus"
                        | "scroll"
                        | "page-pan"
                        | "page-zoom"
                        | "viewport-pan"
                        | "viewport-zoom"
                        | "scene-orbit"
                        | "scene-pan"
                        | "scene-zoom"
                        | "tab-activate"
                        | "tab-close"
                        | "tab-detach"
                        | "tab-drop"
                        | "window-close"
                )
            {
                return RequestProgress::Complete(error_response(
                    StatusCode(400),
                    "unsupported event kind",
                ));
            }
            if let Err(message) = input.validate() {
                return RequestProgress::Complete(error_response(StatusCode(400), message));
            }
            let current = match namespace.read() {
                Ok(current) => current,
                Err(_) => {
                    return RequestProgress::Complete(error_response(
                        StatusCode(500),
                        "namespace lock poisoned",
                    ));
                }
            };
            // Rebase trusted captured native gestures only after BeforeInput
            // has drained the prior refresh. External HTTP remains strict.
            if job.native.is_some()
                && !crate::events::refresh_continuous_input(&mut input, &current)
            {
                return RequestProgress::Complete(error_response(
                    StatusCode(409),
                    "native target was replaced or detached",
                ));
            }
            let frame = if input.window == crate::window::MAIN_WINDOW {
                &current.document.frame
            } else {
                let Some(document) = current.document.windows.get(&input.window) else {
                    return RequestProgress::Complete(error_response(
                        StatusCode(409),
                        "window is no longer active",
                    ));
                };
                &document.frame
            };
            if input.kind == "window-close"
                && frame.window == input.window
                && frame.session == input.session
            {
                let owns = self.repl.as_ref().is_some_and(|(owner, runtime)| {
                    *owner == epoch && runtime.id == input.session && input.session != 0
                });
                let stopped = frame.fault.is_some()
                    || !owns
                    || self
                        .repl
                        .as_ref()
                        .is_some_and(|(_, runtime)| runtime.is_faulted());
                if stopped {
                    drop(current);
                    return RequestProgress::Complete(
                        self.close_inactive_window(&input, epoch, namespace, notify),
                    );
                }
            }
            if let Some(target) = &input.tab_drop {
                let target_frame = if target.window == crate::window::MAIN_WINDOW {
                    Some(&current.document.frame)
                } else {
                    current
                        .document
                        .windows
                        .get(&target.window)
                        .map(|document| &document.frame)
                };
                if target_frame.is_none_or(|frame| {
                    frame.window != target.window
                        || frame.session != input.session
                        || frame.layout_version != target.layout_version
                        || frame.fault.is_some()
                        || target.index > frame.tabs.len()
                }) || self
                    .repl
                    .as_ref()
                    .is_none_or(|(_, runtime)| !runtime.accepts_tab_drop(&input))
                {
                    return RequestProgress::Complete(error_response(
                        StatusCode(409),
                        "drop targets an obsolete or inactive frame",
                    ));
                }
            }
            if frame.window != input.window
                || frame.session != input.session
                || (input.kind != "window-close" && frame.layout_version != input.layout_version)
                || frame.fault.is_some()
                || if input.kind == "window-close"
                    || input.kind.starts_with("developer-open:")
                    || matches!(input.kind.as_str(), "page-pan" | "page-zoom")
                {
                    input.mount != 0
                } else if matches!(
                    input.kind.as_str(),
                    "tab-activate" | "tab-close" | "tab-detach" | "tab-drop"
                ) {
                    !frame.tabs.iter().any(|tab| tab.id == input.mount)
                } else {
                    !frame.mounts.iter().any(|mount| {
                        mount.id == input.mount
                            && if input.kind.starts_with("viewport-") {
                                mount.canvas_viewport
                            } else if input.kind.starts_with("scene-") {
                                mount.scene3d_viewport && mount.camera_controls
                            } else if input.kind == "scroll" {
                                mount.scroll.is_some()
                            } else {
                                mount.interactive
                            }
                    })
                }
                || self.repl.as_ref().is_none_or(|(context, runtime)| {
                    *context != epoch
                        || runtime.id != input.session
                        || (input.kind != "window-close" && !runtime.accepts_window(input.window))
                })
            {
                return RequestProgress::Complete(error_response(
                    StatusCode(409),
                    "event targets an obsolete or inactive frame",
                ));
            }
            if input.kind == "page-zoom" && frame.surface.mode != crate::page::Mode::Canvas
                || input.kind == "page-pan" && frame.surface.mode == crate::page::Mode::Window
            {
                return RequestProgress::Complete(error_response(
                    StatusCode(400),
                    "camera input is unavailable in this Page mode",
                ));
            }
            if matches!(input.kind.as_str(), "viewport-zoom" | "scene-zoom")
                && !frame.mounts.iter().any(|mount| {
                    mount.id == input.mount
                        && input.point.is_some_and(|point| {
                            let point = crate::geometry::Point::new(point[0], point[1]);
                            mount.contains(point)
                                && mount.viewport_clip.is_some_and(|clip| {
                                    mount.view_point(point, true).is_some_and(|point| {
                                        point.x >= clip[0]
                                            && point.y >= clip[1]
                                            && point.x < clip[0] + clip[2]
                                            && point.y < clip[1] + clip[3]
                                    })
                                })
                        })
                })
            {
                return RequestProgress::Complete(error_response(
                    StatusCode(400),
                    "zoom anchor is outside content viewport",
                ));
            }
            if (input.kind.starts_with("viewport-") || input.kind.starts_with("scene-"))
                && input.point.is_none()
                && frame
                    .mounts
                    .iter()
                    .any(|mount| mount.id == input.mount && mount.panel.is_some())
            {
                return RequestProgress::Complete(error_response(
                    StatusCode(400),
                    "world-space UI camera input requires a pointer",
                ));
            }
            Some(input)
        } else {
            None
        };
        if self
            .repl
            .as_ref()
            .is_none_or(|(current, _)| *current != epoch)
        {
            self.next_session = match self.next_session.checked_add(1) {
                Some(id) => id,
                None => {
                    return RequestProgress::Complete(error_response(
                        StatusCode(500),
                        "session identities exhausted",
                    ));
                }
            };
            let viewport = namespace.read().ok().and_then(|current| current.viewport);
            match endor_host::ReplSession::new_with_viewport(self.next_session, viewport) {
                Ok(session) => {
                    if let Ok(mut current) = namespace.write()
                        && current.submission == epoch
                    {
                        current.publish_window_commands(session.id, Vec::new());
                    }
                    self.repl = Some((epoch, session));
                }
                Err(error) => {
                    return RequestProgress::Complete(error_response(
                        StatusCode(500),
                        error.to_string(),
                    ));
                }
            }
        }
        let viewports = match namespace.read() {
            Ok(current) => current_viewports(&current),
            Err(_) => {
                return RequestProgress::Complete(error_response(
                    StatusCode(500),
                    "namespace lock is poisoned",
                ));
            }
        };
        let session = &mut self.repl.as_mut().expect("initialized REPL").1;
        let id = session.id;
        let result = catch_unwind(AssertUnwindSafe(|| -> Result<_> {
            let result = if let Some(event) = &event {
                session.dispatch_incremental(event, &viewports)?
            } else {
                session.execute_incremental(&job.source, &job.namespace, &viewports)?
            };
            let commands = if session.has_pending_refresh() {
                Vec::new()
            } else {
                session.take_window_commands()?
            };
            Ok((result, commands))
        }));
        let faulted = session.is_faulted();
        if session.has_pending_refresh() {
            if let Ok(Ok((result, _))) = result {
                return RequestProgress::Pending(PendingRequest::Input(Box::new(PendingInput {
                    epoch,
                    id,
                    event: event.is_some(),
                    result,
                })));
            }
            // A failed start must never leave a request waiting on a broken VM.
            self.repl = None;
        }
        RequestProgress::Complete(self.finish_input(
            job,
            namespace,
            notify,
            epoch,
            id,
            event.is_some(),
            faulted,
            result,
        ))
    }

    /// 恢复已挂起的请求，保持原 HTTP 响应和运行时身份直到最终完成。
    pub(super) fn resume_request(
        &mut self,
        job: &BuildRequest,
        pending: PendingRequest,
        namespace: &Arc<RwLock<NamespaceState>>,
        notify: &Notify,
    ) -> RequestProgress {
        match pending {
            PendingRequest::BeforeInput => self.begin_request(job, namespace, notify),
            PendingRequest::Input(input) => self.resume_input(job, input, namespace, notify),
            PendingRequest::Candidate(runtime) => {
                self.resume_candidate(job, runtime, namespace, notify)
            }
        }
    }

    /// 继续同一 Console/事件输入的展示刷新，不再次执行其业务回调。
    fn resume_input(
        &mut self,
        job: &BuildRequest,
        mut input: Box<PendingInput>,
        namespace: &Arc<RwLock<NamespaceState>>,
        notify: &Notify,
    ) -> RequestProgress {
        let viewports = match namespace.read() {
            Ok(current)
                if current.submission == input.epoch || current.runtime_session == input.id =>
            {
                current_viewports(&current)
            }
            Ok(_) => {
                self.repl = None;
                return RequestProgress::Complete(superseded(job.submission));
            }
            Err(_) => {
                self.repl = None;
                return RequestProgress::Complete(error_response(
                    StatusCode(500),
                    "namespace lock is poisoned",
                ));
            }
        };
        let Some((epoch, session)) = &mut self.repl else {
            return RequestProgress::Complete(error_response(
                StatusCode(409),
                "input runtime was discarded",
            ));
        };
        if *epoch != input.epoch || session.id != input.id {
            return RequestProgress::Complete(superseded(job.submission));
        }
        let resumed = catch_unwind(AssertUnwindSafe(|| -> Result<_> {
            let document = session.resize_windows_incremental(&viewports, false)?;
            if session.has_pending_refresh() {
                return Ok(None);
            }
            Ok(Some((document, session.take_window_commands()?)))
        }));
        let faulted = session.is_faulted();
        let result = match resumed {
            Ok(Ok(None)) => return RequestProgress::Pending(PendingRequest::Input(input)),
            Ok(Ok(Some((document, commands)))) => {
                input.result.document = document;
                Ok(Ok((input.result, commands)))
            }
            Ok(Err(error)) => Ok(Err(error)),
            Err(panic) => Err(panic),
        };
        RequestProgress::Complete(self.finish_input(
            job,
            namespace,
            notify,
            input.epoch,
            input.id,
            input.event,
            faulted,
            result,
        ))
    }

    #[allow(clippy::too_many_arguments)]
    /// 校验输入仍属于当前域，再一次性发布结果、元数据和原生窗口命令。
    fn finish_input(
        &mut self,
        job: &BuildRequest,
        namespace: &Arc<RwLock<NamespaceState>>,
        notify: &Notify,
        epoch: u64,
        id: u64,
        event: bool,
        faulted: bool,
        result: InputOutcome,
    ) -> HttpResponse {
        let Ok(mut current) = namespace.write() else {
            return error_response(StatusCode(500), "namespace lock is poisoned");
        };
        let epoch = if current.submission != epoch {
            if current.runtime_session != id {
                drop(current);
                self.repl = None;
                return superseded(job.submission);
            }
            // Accepted PUTs invalidate queued inputs, not a mutation already
            // executing on this owner. Settle it before building the candidate
            // so a failed replacement can preserve a consistent live heap.
            if let Some((owner, runtime)) = &mut self.repl
                && runtime.id == id
            {
                *owner = current.submission;
            }
            current.submission
        } else {
            epoch
        };
        match result {
            Ok(Ok((result, commands))) => {
                let projected = !commands.is_empty();
                let rendered = result.document.is_some();
                if let Some(document) = result.document {
                    match current.commit(epoch, document) {
                        Ok(Some(_)) => {}
                        Ok(None) => return superseded(job.submission),
                        Err(error) => return error_response(StatusCode(500), error.to_string()),
                    }
                }
                current.publish_window_commands(id, commands);
                let revision = current.revision;
                drop(current);
                if rendered || projected {
                    notify();
                }
                json(
                    StatusCode(200),
                    serde_json::json!({"ok":true,"execution":if event { "event" } else { "repl" },"persistent":true,"session":id,"input":result.input,"value":result.value,"bindings":result.bindings,"rendered":rendered,"revision":revision}),
                )
            }
            Ok(Err(error)) => {
                if event || faulted {
                    let mut document = current.document.clone();
                    // Preserve the first failure of this VM; subsequent
                    // rejected Console input must not hide the callback cause.
                    // A preserved picture from another generation is unrelated.
                    let fault = (document.frame.session == id)
                        .then(|| document.frame.fault.clone())
                        .flatten()
                        .unwrap_or_else(|| format!("{error:#}").chars().take(1024).collect());
                    document.frame.fault = Some(fault.clone());
                    for view in document.windows.values_mut() {
                        view.frame.fault = Some(fault.clone());
                    }
                    let _ = current.commit(epoch, document);
                    drop(current);
                    notify();
                }
                compile_error(error)
            }
            Err(_) => {
                let mut document = current.document.clone();
                document.frame.fault =
                    Some("Runtime discarded after an internal execution failure".into());
                for view in document.windows.values_mut() {
                    view.frame.fault = document.frame.fault.clone();
                }
                current.runtime_session = 0;
                current.window_commands.clear();
                let _ = current.commit(epoch, document);
                drop(current);
                self.repl = None;
                notify();
                json(
                    StatusCode(500),
                    serde_json::json!({"ok":false,"session_reset":true,"error":"Console execution failed internally; session discarded, displayed page preserved"}),
                )
            }
        }
    }

    /// An accepted PUT changes the submission, but does not revoke the old
    /// runtime. DELETE clears runtime_session and does revoke it.
    /// 接受较新提交不等于立即销毁旧 VM；候选失败时仍需保留当前可用运行时。
    pub(super) fn preserve_live_owner(&mut self, current: &NamespaceState) {
        if let Some((epoch, runtime)) = &mut self.repl {
            if current.runtime_session == runtime.id {
                *epoch = current.submission;
            } else {
                self.repl = None;
            }
        }
    }

    /// 为完整 PUT 建立隔离候选，保留旧运行时直到新候选完整成功。
    fn begin_program_request(
        &mut self,
        job: &BuildRequest,
        namespace: &Arc<RwLock<NamespaceState>>,
        notify: &Notify,
    ) -> RequestProgress {
        let viewport = match namespace.read() {
            Ok(current) if current.submission == job.submission => current.viewport,
            Ok(current) => {
                self.preserve_live_owner(&current);
                return RequestProgress::Complete(superseded(job.submission));
            }
            Err(_) => {
                return RequestProgress::Complete(error_response(
                    StatusCode(500),
                    "namespace lock poisoned",
                ));
            }
        };
        self.next_session = match self.next_session.checked_add(1) {
            Some(id) => id,
            None => {
                return RequestProgress::Complete(error_response(
                    StatusCode(500),
                    "session identities exhausted",
                ));
            }
        };
        let id = self.next_session;
        let started = catch_unwind(AssertUnwindSafe(|| {
            endor_host::ReplSession::load_incremental(
                &job.source,
                &job.namespace,
                job.submission,
                id,
                viewport,
            )
        }));
        let result = match started {
            Ok(Ok((runtime, None))) => {
                return RequestProgress::Pending(PendingRequest::Candidate(Box::new(runtime)));
            }
            Ok(Ok((runtime, Some(document)))) => match runtime.take_window_commands() {
                Ok(commands) => Ok(Ok((runtime, document, commands))),
                Err(error) => Ok(Err(error)),
            },
            Ok(Err(error)) => Ok(Err(error)),
            Err(panic) => Err(panic),
        };
        RequestProgress::Complete(self.finish_candidate(job, namespace, notify, id, result))
    }

    /// 推进候选的布局与视口收敛，检查其是否已被新提交撤销。
    fn resume_candidate(
        &mut self,
        job: &BuildRequest,
        mut runtime: Box<endor_host::ReplSession>,
        namespace: &Arc<RwLock<NamespaceState>>,
        notify: &Notify,
    ) -> RequestProgress {
        let viewports = match namespace.read() {
            Ok(current) if current.submission == job.submission => {
                // Secondary native resources still belong to the old VM. Only
                // the primary shell's current viewport applies to a candidate.
                current
                    .viewport
                    .map(|size| (crate::window::MAIN_WINDOW, size))
                    .into_iter()
                    .collect()
            }
            Ok(current) => {
                self.preserve_live_owner(&current);
                return RequestProgress::Complete(superseded(job.submission));
            }
            Err(_) => {
                return RequestProgress::Complete(error_response(
                    StatusCode(500),
                    "namespace lock poisoned",
                ));
            }
        };
        let id = runtime.id;
        let resumed = catch_unwind(AssertUnwindSafe(|| -> Result<_> {
            let document = runtime.resize_windows_incremental(&viewports, false)?;
            if runtime.has_pending_refresh() {
                return Ok(None);
            }
            let document =
                document.ok_or_else(|| anyhow::anyhow!("program did not publish a Page"))?;
            Ok(Some((document, runtime.take_window_commands()?)))
        }));
        let result = match resumed {
            Ok(Ok(None)) => return RequestProgress::Pending(PendingRequest::Candidate(runtime)),
            Ok(Ok(Some((document, commands)))) => Ok(Ok((*runtime, document, commands))),
            Ok(Err(error)) => Ok(Err(error)),
            Err(panic) => Err(panic),
        };
        RequestProgress::Complete(self.finish_candidate(job, namespace, notify, id, result))
    }

    /// 候选成功才切换活跃 VM 和文档；失败保留先前已发布程序。
    fn finish_candidate(
        &mut self,
        job: &BuildRequest,
        namespace: &Arc<RwLock<NamespaceState>>,
        notify: &Notify,
        id: u64,
        result: CandidateOutcome,
    ) -> HttpResponse {
        let Ok(mut current) = namespace.write() else {
            return error_response(StatusCode(500), "namespace lock poisoned");
        };
        if current.submission != job.submission {
            self.preserve_live_owner(&current);
            return superseded(job.submission);
        }
        match result {
            Ok(Ok((runtime, document, commands))) => match current.commit(job.submission, document)
            {
                Ok(Some(revision)) => {
                    current.publish_window_commands(id, commands);
                    self.repl = Some((job.submission, runtime));
                    drop(current);
                    notify();
                    json(
                        StatusCode(200),
                        serde_json::json!({"ok":true,"namespace":job.namespace,"submission":job.submission,"revision":revision,"diagnostics":[],"session":id,"persistent":true}),
                    )
                }
                Ok(None) => superseded(job.submission),
                Err(error) => error_response(StatusCode(500), error.to_string()),
            },
            failed => {
                // The candidate never mutated the displayed program's heap.
                // Restore that program's event epoch when a replacement fails.
                self.preserve_live_owner(&current);
                drop(current);
                match failed {
                    Ok(Err(error)) => compile_error(error),
                    _ => error_response(
                        StatusCode(500),
                        "candidate failed internally; previous runtime preserved",
                    ),
                }
            }
        }
    }
}

/// 处理隔离表达式或一次性文档编译；编译前后核对 submission，只发布仍有效的完整快照。
fn build(
    job: &BuildRequest,
    namespace: &Arc<RwLock<NamespaceState>>,
    notify: &Notify,
) -> Response<std::io::Cursor<Vec<u8>>> {
    if job.kind == RequestKind::Expression {
        return match catch_unwind(AssertUnwindSafe(|| endor_host::evaluate(&job.source))) {
            Ok(Ok(value)) => json(
                StatusCode(200),
                serde_json::json!({"ok":true,"value":value,"execution":"isolated-expression","persistent":false}),
            ),
            Ok(Err(error)) => compile_error(error),
            Err(_) => error_response(StatusCode(500), "expression execution failed internally"),
        };
    }
    match namespace.read() {
        Ok(current) if current.submission != job.submission => return superseded(job.submission),
        Err(_) => return error_response(StatusCode(500), "namespace lock is poisoned"),
        _ => {}
    }
    // Only an independent native snapshot crosses into the published state.
    // A fresh VM is discarded on success, error and unwind alike.
    let result = catch_unwind(AssertUnwindSafe(|| {
        endor_host::compile(&job.source, &job.namespace, job.submission)
    }));
    let Ok(mut current) = namespace.write() else {
        return error_response(StatusCode(500), "namespace lock is poisoned");
    };
    if current.submission != job.submission {
        return superseded(job.submission);
    }
    match result {
        Ok(Ok(document)) => match current.commit(job.submission, document) {
            Ok(Some(revision)) => {
                drop(current);
                notify();
                json(
                    StatusCode(200),
                    serde_json::json!({ "ok": true, "namespace": job.namespace,
                    "submission": job.submission, "revision": revision, "diagnostics": [] }),
                )
            }
            Ok(None) => superseded(job.submission),
            Err(error) => error_response(StatusCode(500), error.to_string()),
        },
        Ok(Err(error)) => compile_error(error),
        Err(_) => error_response(
            StatusCode(500),
            "candidate execution failed internally; previous document preserved",
        ),
    }
}