//! 唯一串行执行 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>>>;
// 外层 thread::Result 区分 Rust panic，内层 Result 区分可诊断编译/运行错误；正常结果另带待发布窗口命令。
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 {
    // 输入开始时所属 submission，与 id（VM 会话）分开；新 PUT 接受时可保留正在执行的原 VM 至输入收敛。
    epoch: u64,
    id: u64,
    event: bool,
    result: endor_host::ReplResult,
}

/// 挂起 HTTP 请求的阶段：等旧刷新、继续输入或继续隔离候选。
pub(super) enum PendingRequest {
    // 仅等待旧只读刷新结束，业务输入尚未执行；恢复时可重新走入口，而 Input 阶段不能重放业务副作用。
    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();
    // 主窗口逻辑视口优先覆盖 map 同名值，主壳可以跨会话沿用；副窗口必须仍属于当前发布世代。
    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,
            };
        };
        // 正常域允许有界关闭回调，故障域只做宿主清理；catch_unwind 防止清理 panic 跳过原生退出通知。
        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>>> {
        // 故障/失活资源的专用关闭只处理副窗口，主壳由 OS 关闭控制走全应用退出流程。
        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)
            }));
            // 退役失败不能继续借用半清理 VM，放弃整个活域并在下方清其他帧的会话与命令资格。
            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");
        };
        // 即便副窗口数值 ID 相同，也不能让旧会话的关闭请求删除新会话投影。
        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;
        };
        // submission 变化先尝试保留仍权威的活域所有者，不立即拿旧 epoch 继续发布后台结果。
        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)?
            };
            // 任务后续刷新未收敛时可以发布较早已完整的 resize 文档，但暂扣任务产生的命令，防止半完成副作用外泄。
            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,
        };
        // 耗时 VM 工作期间 HTTP 线程可能接受 PUT/DELETE，发布写锁内再核对 submission 与 session 双身份。
        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 {
                    // 无新文档也可能有标题/焦点等原生命令；只在实际有投影变化时唤醒 GUI，避免空闲无效重绘。
                    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(),
                    // 内部 panic 与普通背景错误不同，必须丢弃 VM 并撤销所有原生命令，不能继续运行损坏的域。
                    _ => {
                        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 {
        // 完整 PUT 独立候选，表达式隔离求值；只有 REPL/事件会进入当前持久域并等待已有刷新。
        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));
        }
        // 重置丢弃内存会话但保留显示文档，清帧 session/focus 使旧画面不再接受旧对象事件。
        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);
        }
        // 事件先做 JSON 结构、kind 白名单和 Input 字段校验，再用权威发布帧核对具体资源能力。
        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
            };
            // 窗口关闭允许处理已失活/故障副窗口，不像点击一样要求可运行 VM；由专用宿主退役路径处理。
            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",
                    ));
                }
            }
            // 统一事件归属门：窗口/session/版本及操作对应资源必须都合法；window-close 特别允许旧布局版本。
            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"
                ) {
                    // 标签操作的 mount 字段承载 Tab ID，而普通输入承载 Mount ID；两套身份不能混着查询。
                    !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-") {
                                // 3D 手势还要求显式控制器能力，scroll 要有滚动投影，点击/焦点则要求 interactive。
                                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",
                ));
            }
            // 兼容页面仅 Canvas 可缩放，Window 模式不接受 pan；结构合法不等于当前页面模式支持操作。
            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",
                ));
            }
            // 缩放锚点须经过面板反投影与内容 clip 检查，不能落在视口边框或被遮挡面板上仍缩放。
            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",
                ));
            }
            // 世界 UI 的非线性位移变换需要位置锚点；普通二维视口的线性位移可以省略 point。
            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)
        {
            // 没有当前 epoch 会话时惰性建立空 REPL，每次分配新会话 ID 并检查溢出，禁止复用旧输入身份。
            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)?
            };
            // 业务执行可能已改 VM，但展示未完成时暂不取窗口命令；PendingInput 保存结果并只恢复展示阶段。
            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,
                })));
            }
            // 启动失败且残留挂起刷新时直接丢弃域，不能保留一个无法安全推进的 HTTP 请求。
            // 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() {
            // 正在执行输入只要仍属当前提交或仍拥有权威 VM 就可收敛；两项皆失效才彻底废弃。
            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<_> {
            // 这里只恢复 pending 展示/尺寸工作，不再调用 execute/dispatch，所以前一个量子的源副作用不会重放。
            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");
        };
        // 接受新 PUT 会使排队旧输入失效，但已执行的活域输入需先收敛，才能在候选失败时留下内部一致的旧程序。
        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();
                // rendered 只表示此次交付带新文档，成功执行但页面未变仍可返回 value/session；不因此虚增 revision。
                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.
                    // 同 VM 保留最初故障原因，后续“域已停止”错误不能覆盖真正回调故障；其他世代旧图不沿用故障。
                    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();
                }
                // 输入 panic 保留最近画面但撤销运行时/命令并发布故障条，再将 self.repl 丢弃。
                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;
        // 候选装载运行在独立 ReplSession，未成功前不修改 self.repl，旧程序仍是权威展示来源。
        let started = catch_unwind(AssertUnwindSafe(|| {
            endor_host::ReplSession::load_incremental(
                &job.source,
                &job.namespace,
                job.submission,
                id,
                viewport,
            )
        }));
        let result = match started {
            // 候选业务装载成功但布局未完成，整个候选移进 PendingRequest；不能把 None 当成空文档提交。
            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.
                // 候选不沿用旧副窗口的视口；副窗口资源归旧 VM，只有主壳当前尺寸可作为新候选分配条件。
                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);
            }
            // 所有刷新结束仍没有 Page 文档时明确失败，不用旧显示值冒充候选自己的输出。
            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)
            {
                // 写锁内先原子提交文档，再切换命令世代与活域；解锁后唤醒 GUI，新 VM 至此才取代旧 VM。
                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.
                // 候选失败只更新旧活域 epoch，保留其堆与页面；不存在旧域时也不会借用其他会话伪造成功。
                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>>> {
    // 隔离表达式不读写持久 REPL 堆，也不发布文档，响应明确 persistent=false。
    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.
    // 一次性构建同样用独立 VM，计算锁外执行，返回后再次核对 submission 才能发布原生快照。
    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",
        ),
    }
}