//! 本机 HTTP API 入口:路由请求、限制输入大小、观察活动,并将执行工作送入串行 worker。 //! HTTP 线程不操作 VM;共享 NamespaceState 只保存最近完整发布文档和资源命令。提交序号用来拒绝已被较新 PUT/DELETE 取代的结果。 //! 关闭流程先请求运行时有界清理,再通过完成通知让原生事件循环退出。 //! mod guide; mod worker; use guide::agent_guide; use worker::{PendingRequest, RequestProgress, Worker}; use std::{ io::Read, ops::{Deref, DerefMut}, panic::{AssertUnwindSafe, catch_unwind}, sync::{ Arc, OnceLock, RwLock, mpsc::{Receiver, SyncSender, TrySendError, sync_channel}, }, sync::{ Condvar, Mutex, atomic::{AtomicBool, Ordering}, }, thread, }; use anyhow::{Context as _, Result}; use tiny_http::{Header, Method, Request, Response, Server, StatusCode}; use winit::event_loop::EventLoopProxy; use crate::{ activity::{Activity, ActivityLog}, app::AppEvent, document::NamespaceState, endor_host, }; // 只监听 IPv4 回环固定端口,接口用于本进程原生输入和本机 agent,不是对外网络服务器。 pub const API_ADDRESS: &str = "127.0.0.1:4737"; const MAX_CODE_SIZE: u64 = endor_host::MAX_SOURCE_BYTES as u64; type Notify = Arc; #[derive(Clone, Debug)] /// 运行时清理的摘要与可选错误,供最终原生退出结果报告。 pub struct ShutdownOutcome { pub summary: String, pub error: Option, } type ShutdownCompletion = Arc<(Mutex>, Condvar)>; #[derive(Clone)] /// 跨线程的退出控制句柄,使用有界通知和完成条件变量,不携带 VM。 pub struct RuntimeControl { sender: SyncSender<()>, requested: Arc, completion: ShutdownCompletion, native_sender: Arc>>, namespace: Arc>, } impl RuntimeControl { /// Only the native input relay calls this blocking acknowledgement path. /// Window threads still use a bounded, nonblocking Input sender. pub(crate) fn dispatch_native(&self, input: crate::events::Input) -> Result<()> { anyhow::ensure!(!self.is_exiting(), "application is exiting"); let (done, completed) = sync_channel(1); let job = { let state = self .namespace .read() .map_err(|_| anyhow::anyhow!("namespace lock is poisoned"))?; BuildRequest { request: RequestReply::Native(done), kind: RequestKind::Event, source: String::new(), namespace: state.name.clone(), submission: state.submission, native: Some(input), } }; self.native_sender .get() .ok_or_else(|| anyhow::anyhow!("native input queue is unavailable"))? .send(job) .map_err(|_| anyhow::anyhow!("native input worker stopped"))?; completed .recv() .map_err(|_| anyhow::anyhow!("native input acknowledgement stopped"))? .map_err(anyhow::Error::msg) } /// 只发出一次退出请求,唤醒 worker 执行清理,并通过完成句柄通知原生循环。 pub fn request_shutdown(&self) -> bool { // 原子标志合并重复退出请求;已请求直接返回 true,不在 GUI 线程重复排队清理。 if self.requested.swap(true, Ordering::AcqRel) { return true; } if self.sender.try_send(()).is_ok() { true } else { // 控制队列不可用时回滚请求标志,允许调用方识别失败并再次尝试,而非永远卡在假退出状态。 self.requested.store(false, Ordering::Release); false } } /// 读取退出标志,使窗口事件和新输入尽早停止进入运行时。 pub fn is_exiting(&self) -> bool { self.requested.load(Ordering::Acquire) } /// 在明确超时内等待 worker 的清理结果,不能无限阻塞退出。 pub fn wait_shutdown(&self, timeout: std::time::Duration) -> Option { let (lock, ready) = &*self.completion; let result = lock.lock().ok()?; let (result, _) = ready // 条件变量处理虚假唤醒,直到完成槽非空或超时;返回 None 也可能来自锁中毒,不能当作清理成功。 .wait_timeout_while(result, timeout, |result| result.is_none()) .ok()?; result.clone() } } /// worker 持有的关闭接收端、控制状态和 GUI 完成通知代理。 struct Shutdown { receiver: Receiver<()>, control: RuntimeControl, proxy: EventLoopProxy, } #[derive(Clone, Copy, Debug, PartialEq, Eq)] /// 送入执行队列的请求类别,区分完整源码、持久输入、事件与重置。 enum RequestKind { Event, Code, Expression, Repl, ResetRepl, } /// 保留原 HTTP 请求与提交身份的执行工作,完成或失败后由 worker 回应。 struct BuildRequest { request: RequestReply, native: Option, kind: RequestKind, source: String, namespace: String, submission: u64, } /// 绑定回环 API,启动请求接收与单 VM worker,返回活动日志及一次性退出控制句柄。 pub fn start( namespace: Arc>, proxy: EventLoopProxy, ) -> Result<(Activity, RuntimeControl)> { // 退出控制仅需一个消息槽,正常执行请求另走容量8队列,避免工作排满后无法通知关闭。 let (sender, receiver) = sync_channel(1); let control = RuntimeControl { sender, requested: Arc::new(AtomicBool::new(false)), completion: Arc::new((Mutex::new(None), Condvar::new())), native_sender: Arc::new(OnceLock::new()), namespace: namespace.clone(), }; let shutdown = Shutdown { receiver, control: control.clone(), proxy: proxy.clone(), }; // 绑定失败在启动阶段反馈,不能假装本机 API 可用;后台通知仅发送 GUI 事件,不直接画窗口。 let server = Server::http(API_ADDRESS) .map_err(|error| anyhow::anyhow!(error)) .with_context(|| format!("bind Tessera API to {API_ADDRESS}"))?; let notify: Notify = Arc::new(move || { let _ = proxy.send_event(AppEvent::DocumentUpdated); }); let activity = Arc::new(Mutex::new(ActivityLog::default())); start_server_observed( Arc::new(server), namespace, notify.clone(), activity.clone(), notify, Some(shutdown), Some(control.native_sender.clone()), )?; Ok((activity, control)) } #[cfg(test)] /// 为本地监听服务建立请求处理线程,并把执行请求转交有界队列。 fn start_server( server: Arc, namespace: Arc>, notify: Notify, ) -> Result> { start_server_observed( server, namespace, notify, Arc::new(Mutex::new(ActivityLog::default())), Arc::new(|| {}), None, None, ) } /// 启动带活动观察的 HTTP 服务,将请求状态变化通知管理界面。 fn start_server_observed( server: Arc, namespace: Arc>, notify: Notify, activity: Activity, observer_notify: Notify, shutdown: Option, native_port: Option>>>, ) -> Result> { // HTTP 接收与 VM 执行分离,最多8个排队请求;源代码只在单个 endor-worker 中操作 VM。 let (sender, receiver) = sync_channel::(8); if let Some(native_port) = native_port { native_port .set(sender.clone()) .map_err(|_| anyhow::anyhow!("native input queue already attached"))?; } let worker_namespace = namespace.clone(); let worker_notify = notify.clone(); let exiting = shutdown .as_ref() .map(|shutdown| shutdown.control.requested.clone()); thread::Builder::new() .name("tessera-endor-worker".into()) .spawn(move || { let mut worker = Worker::default(); // 分片请求跨循环保留原 HTTP 请求和继续句柄;同一时刻只推进一个有未完成 VM 执行的输入。 let mut pending: Option<(BuildRequest, PendingRequest)> = None; loop { if let Some(shutdown) = &shutdown && shutdown.receiver.try_recv().is_ok() { // 退出优先于下一量子:先撤去待响应请求,再由 shutdown_runtime 清理 VM 续体与生命周期资源。 if let Some((job, input)) = pending.take() { drop(input); let _ = job .request .respond(error_response(StatusCode(503), "application is exiting")); } let outcome = worker.shutdown_runtime(); // 排队请求全部明确回应503,不能留着客户端无期限等待已经停止的执行线程。 while let Ok(job) = receiver.try_recv() { let _ = job .request .respond(error_response(StatusCode(503), "application is exiting")); } if let Ok(mut current) = worker_namespace.write() { // 撤销所有帧的会话/焦点与原生命令,让 GUI 不再把旧资源当成仍可执行的活域。 current.runtime_session = 0; current.window_commands.clear(); current.document.frame.session = 0; current.document.frame.focus = None; for view in current.document.windows.values_mut() { view.frame.session = 0; view.frame.focus = None; } } eprintln!("{}", outcome.summary); let (lock, ready) = &*shutdown.control.completion; if let Ok(mut result) = lock.lock() { *result = Some(outcome.clone()); // 先在锁内保存清理结果再唤醒等待者,同时用 AppEvent 通知原生事件循环退出,两条路径复用同一 outcome。 ready.notify_all(); } let _ = shutdown.proxy.send_event(AppEvent::ShutdownComplete { error: outcome.error, }); break; } // 分片只让出线程,不并行执行后续输入;当前请求完成才回应并处理下一队列项。 if let Some((job, input)) = pending.take() { match worker.resume_request(&job, input, &worker_namespace, &worker_notify) { RequestProgress::Pending(input) => { pending = Some((job, input)); // A soft VM quantum is a control boundary, not a // minimum-duration time slice. Yield to other ready // threads without charging a timer delay per quantum. thread::yield_now(); } RequestProgress::Complete(response) => { let _ = job.request.respond(response); } } continue; } // 空闲输入边界先推进视口刷新与任务,待执行刷新/协作任务时零等待,否则最多100ms检查一次。 worker.refresh_viewport(&worker_namespace, &worker_notify); let wait = if worker.repl.as_ref().is_some_and(|(_, runtime)| { runtime.has_pending_refresh() || runtime.has_cooperative_tasks() }) { std::time::Duration::ZERO } else { std::time::Duration::from_millis(100) }; match receiver.recv_timeout(wait) { Ok(job) => { match worker.begin_request(&job, &worker_namespace, &worker_notify) { RequestProgress::Pending(input) => { if job.native.is_none() && matches!(&input, PendingRequest::BeforeInput) { eprintln!( "Tessera worker: submission {} waits for active refresh", job.submission ); } pending = Some((job, input)); } RequestProgress::Complete(response) => { let _ = job.request.respond(response); } } } Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break, Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { if worker.repl.as_ref().is_some_and(|(epoch, runtime)| { worker_namespace.read().is_ok_and(|current| { current.submission != *epoch && current.runtime_session != runtime.id }) }) { // 当前持久域同时失去 submission 和 runtime_session 归属时退役;不能仅因源码请求排队就误删仍权威的旧域。 worker.repl = None; } if wait.is_zero() { thread::yield_now(); } } } } }) .context("start Aster execution worker")?; thread::Builder::new() .name("tessera-agent-api".to_owned()) .spawn(move || { for request in server.incoming_requests() { if exiting .as_ref() .is_some_and(|flag| flag.load(Ordering::Acquire)) // 退出后拒绝写入/执行路由,GET 与 OPTIONS 仍可读取最近发布值或探测接口。 && !matches!(request.method(), Method::Get | Method::Options) { let _ = request.respond(error_response(StatusCode(503), "application is exiting")); continue; } handle( request, &namespace, ¬ify, &sender, &activity, &observer_notify, ); } }) .context("start Tessera agent API thread") } // Capture only allowlisted headers: credentials and cookies never enter the inspector. /// HTTP 请求的观察包装,在读取正文和响应时维护 Network 工具记录。 /// HTTP and in-process native jobs share one serial execution/refresh path. /// Native envelopes cannot be created by an HTTP header or request body. enum RequestReply { Http(ObservedRequest), Native(SyncSender>), } impl RequestReply { fn respond(self, response: Response>>) -> std::io::Result<()> { match self { Self::Http(request) => request.respond(response), Self::Native(done) => { let status = response.status_code().0; let result = if status == 200 || status == 409 { Ok(()) } else { Err(String::from_utf8_lossy(response.into_reader().get_ref()) .chars() .take(2048) .collect()) }; let _ = done.send(result); Ok(()) } } } } struct ObservedRequest { inner: Request, activity: Activity, id: u64, notify: Notify, } impl ObservedRequest { /// 从请求提取允许观察的头部与显示来源,创建活动记录并通知工具界面。 fn new(inner: Request, activity: Activity, notify: Notify) -> Self { // 活动观察只保留 Content-Type/Length;X-Tessera-Origin 仅判来源标签,不把任意认证头复制进工具日志。 let headers = inner .headers() .iter() .filter(|h| h.field.equiv("Content-Type") || h.field.equiv("Content-Length")) .map(ToString::to_string) .collect::>() .join("\n"); // console 标记只是显示来源,不提供权限,也不改变 VM 执行语义。 let origin = if inner .headers() .iter() .any(|h| h.field.equiv("X-Tessera-Origin") && h.value.as_str() == "console") { "Console" } else { "HTTP" } .to_owned(); let id = activity.lock().unwrap().begin( inner.method().to_string(), inner.url().to_owned(), origin, headers, ); notify(); Self { inner, activity, id, notify, } } /// 将受限请求正文记录到活动预览,不在观察器中执行源码。 fn record_body(&self, body: &str) { self.activity.lock().unwrap().body(self.id, body); (self.notify)(); } /// 发送最终 HTTP 响应,并同步结束该条活动记录。 fn respond(self, response: Response>>) -> std::io::Result<()> { let status = response.status_code(); let headers = response.headers().to_vec(); let length = response.data_length(); // 响应正文本已在内存 Cursor,记录预览后用原 status/headers/reader/length 重建响应,不消费正文内容。 let reader = response.into_reader(); // 记录的是已构造响应,即使后面 socket 发送失败也已结束活动记录,不能据此声称客户端成功接收。 self.activity.lock().unwrap().finish( self.id, status.0, &String::from_utf8_lossy(reader.get_ref()), headers .iter() .map(ToString::to_string) .collect::>() .join("\n"), ); (self.notify)(); self.inner .respond(Response::new(status, headers, reader, length, None)) } } impl Deref for ObservedRequest { type Target = Request; /// 将包装类型的只读访问转交给内部当前上下文。 fn deref(&self) -> &Request { &self.inner } } impl DerefMut for ObservedRequest { /// 将包装类型的可变访问转交给内部当前上下文。 fn deref_mut(&mut self) -> &mut Request { &mut self.inner } } /// 解析 HTTP 方法和路径;只读路由读取发布值,执行路由交给 worker。 fn handle( request: Request, namespace: &Arc>, notify: &Notify, sender: &SyncSender, activity: &Activity, observer_notify: &Notify, ) { let request = ObservedRequest::new(request, activity.clone(), observer_notify.clone()); let method = request.method().clone(); // 路由匹配去掉查询串,namespace 从固定路径提取;不把 query 当作命名空间或源代码。 let path = request .url() .split('?') .next() .unwrap_or(request.url()) .to_owned(); let evaluation = namespace_from_route(&path, "/evaluate"); let repl = namespace_from_route(&path, "/repl"); let event = namespace_from_route(&path, "/events"); // PUT/code 完整替换,POST 在 events/repl/evaluate 中分类,DELETE/repl 是持久会话重置;其他交给资源读取/清空路径。 let route = if method == Method::Put { namespace_from_path(&path).map(|name| (name, RequestKind::Code)) } else if method == Method::Post { event .map(|name| (name, RequestKind::Event)) .or_else(|| repl.map(|name| (name, RequestKind::Repl))) .or(evaluation.map(|name| (name, RequestKind::Expression))) } else if method == Method::Delete { repl.map(|name| (name, RequestKind::ResetRepl)) } else { None }; if let Some((addressed, kind)) = route { submit(request, addressed, namespace, sender, kind); return; } let response = if method == Method::Get && path == "/agent" { match namespace.read() { Ok(namespace) => markdown(StatusCode(200), agent_guide(&namespace)), Err(_) => error_response(StatusCode(500), "namespace lock is poisoned"), } } else if let Some(addressed_namespace) = namespace_from_path(&path).map(str::to_owned) { code_request(method, &addressed_namespace, namespace, notify) } else if method == Method::Options { text(StatusCode(204), "", "text/plain") } else { error_response(StatusCode(404), "route not found; start with GET /agent") }; let _ = request.respond(response); } /// 处理源码资源的 GET/DELETE 等请求,返回对应版本头或撤销运行时所有权。 fn code_request( method: Method, addressed_namespace: &str, namespace: &Arc>, notify: &Notify, ) -> Response>> { let Ok(current) = namespace.read() else { return error_response(StatusCode(500), "namespace lock is poisoned"); }; if addressed_namespace != current.name { return error_response( StatusCode(404), format!( "namespace `{addressed_namespace}` is not active; current namespace is `{}`", current.name ), ); } // GET 返回最近成功发布的 source 而非正在排队的输入,并通过头部暴露 revision/session/window/layout 及逻辑视口。 if method == Method::Get { let mut response = text( StatusCode(200), ¤t.document.source, "text/plain; charset=utf-8", ); add_header( &mut response, "X-Tessera-Revision", ¤t.revision.to_string(), ); add_header( &mut response, "X-Tessera-Session", ¤t.document.frame.session.to_string(), ); add_header( &mut response, "X-Tessera-Window", ¤t.document.frame.window.to_string(), ); add_header( &mut response, "X-Tessera-Layout-Version", ¤t.document.frame.layout_version.to_string(), ); if let Some(viewport) = current.viewport { add_header( &mut response, "X-Tessera-Viewport-Width", &viewport.width.to_string(), ); add_header( &mut response, "X-Tessera-Viewport-Height", &viewport.height.to_string(), ); } add_header( &mut response, "X-Tessera-Source-Kind", if current .document .root .as_ref() .is_some_and(|root| root.kind == "repl-session") { "repl-transcript" } else { "source-snapshot" }, ); return response; } // 释放读锁后才申请写锁,避免同一线程持读锁升级为写锁死锁;修改结束也先解锁再发送 GUI 通知。 drop(current); match method { Method::Delete => match namespace.write() { Ok(mut current) => { let revision = match current.clear() { Ok(revision) => revision, Err(error) => return error_response(StatusCode(500), error.to_string()), }; drop(current); notify(); json( StatusCode(200), serde_json::json!({ "ok": true, "namespace": addressed_namespace, "revision": revision }), ) } Err(_) => error_response(StatusCode(500), "namespace lock is poisoned"), }, Method::Options => text(StatusCode(204), "", "text/plain"), _ => error_response(StatusCode(405), "code supports GET, PUT, and DELETE"), } } /// 读取并检查请求体,分配提交身份,再把工作送进有界执行队列。 fn submit( mut request: ObservedRequest, addressed: &str, namespace: &Arc>, sender: &SyncSender, kind: RequestKind, ) { let source = match read_code(&mut request) { Ok(source) => source, Err(error) => { let _ = request.respond(error_response(StatusCode(400), error.to_string())); return; } }; // 读取成功后才保存受限活动正文预览;超长或非法 UTF-8 请求不会进入执行队列。 request.record_body(&source); let Ok(mut current) = namespace.write() else { let _ = request.respond(error_response( StatusCode(500), "namespace lock is poisoned", )); return; }; if current.name != addressed { let _ = request.respond(error_response(StatusCode(404), "namespace is not active")); return; } // 仅完整 Code 请求分配新 submission,事件/REPL/表达式继承当前身份,保留共享域输入顺序。 let submission = match if kind == RequestKind::Code { current.next_submission() } else { Ok(current.submission) } { Ok(id) => id, Err(error) => { let _ = request.respond(error_response(StatusCode(500), error.to_string())); return; } }; let job = BuildRequest { request: RequestReply::Http(request), native: None, kind, source, namespace: addressed.into(), submission, }; // 持 namespace 写锁完成入队和接受身份更新;worker 提交使用同一锁,保证接受与发布之间没有竞态窗口。 match sender.try_send(job) { // A rejected queue entry never invalidates accepted work. The worker // uses this same lock when committing, so publication is atomic. Ok(()) => { if kind == RequestKind::Code { current.submission = submission; } } // 队列拒绝时不推进 submission,所以已经接受的旧工作不会因为本次失败请求而失去发布资格。 Err(TrySendError::Full(job)) => { let _ = job .request .respond(error_response(StatusCode(503), "build queue is full")); } Err(TrySendError::Disconnected(job)) => { let _ = job.request.respond(error_response( StatusCode(503), "execution worker is unavailable", )); } } } /// 生成较旧请求已被新提交取代的响应,拒绝晚到结果覆盖当前文档。 fn superseded(submission: u64) -> Response>> { json( StatusCode(409), serde_json::json!({ "ok": false, "submission": submission, "error": "submission superseded by a newer PUT or DELETE" }), ) } /// 从固定路由格式提取命名空间,避免把不匹配的路径当作代码资源。 fn namespace_from_route<'a>(path: &'a str, suffix: &str) -> Option<&'a str> { // 同时检查前后缀且命名空间非空、无额外斜杠;不做模糊路径前缀匹配。 path.strip_prefix("/v1/namespaces/")? .strip_suffix(suffix) .filter(|namespace| !namespace.is_empty() && !namespace.contains('/')) } /// 从固定路由格式提取命名空间,避免把不匹配的路径当作代码资源。 fn namespace_from_path(path: &str) -> Option<&str> { namespace_from_route(path, "/code") } /// 按源码字节上限读取 UTF-8 请求体,读取失败或超限时返回诊断。 fn read_code(request: &mut Request) -> Result { if request .body_length() .is_some_and(|length| length as u64 > MAX_CODE_SIZE) { anyhow::bail!("code exceeds the 1 MiB limit"); } // 已知 Content-Length 先预拒绝,再在流上最多读上限+1;防止缺失/不实长度绕过容量限制。 // read_to_string 要求 UTF-8,不用有损替换改变用户源码。 let mut code = String::new(); request .as_reader() .take(MAX_CODE_SIZE + 1) .read_to_string(&mut code) .context("read code")?; if code.len() as u64 > MAX_CODE_SIZE { anyhow::bail!("code exceeds the 1 MiB limit"); } Ok(code) } /// 将前端/编译/执行诊断编码为 API 错误响应,保留用户可理解的位置和消息。 fn compile_error(error: anyhow::Error) -> Response>> { let mut diagnostic = serde_json::json!({ "severity": "error", "message": format!("{error:#}") }); // 有 SourceDiagnostic 才附行/列和半开字节范围;其他运行时错误仍保留完整错误链消息,不编造源码位置。 if let Some(source) = error.downcast_ref::() { diagnostic["line"] = source.location.line.into(); diagnostic["column"] = source.location.column.into(); diagnostic["start_byte"] = source.diagnostic.span.start.into(); diagnostic["end_byte"] = source.diagnostic.span.end.into(); } json( StatusCode(422), serde_json::json!({ "ok": false, "diagnostics": [diagnostic] }), ) } /// 生成统一 JSON 错误响应,使调用方能区分协议错误与成功结果。 fn error_response( status: StatusCode, message: impl Into, ) -> Response>> { json(status, serde_json::json!({ "error": message.into() })) } /// 以明确的内容类型和状态码构造响应,并应用本机 API 的公共响应头。 fn json(status: StatusCode, value: serde_json::Value) -> Response>> { text( status, &value.to_string(), "application/json; charset=utf-8", ) } /// 以明确的内容类型和状态码构造响应,并应用本机 API 的公共响应头。 fn markdown(status: StatusCode, value: String) -> Response>> { text(status, &value, "text/markdown; charset=utf-8") } /// 以明确的内容类型和状态码构造响应,并应用本机 API 的公共响应头。 fn text(status: StatusCode, value: &str, content_type: &str) -> Response>> { // 公共响应头只允许固定本机 Origin/方法/头名,非通配任意来源;这些头不替代回环绑定与路由校验。 let mut response = Response::from_string(value).with_status_code(status); add_header(&mut response, "Content-Type", content_type); add_header( &mut response, "Access-Control-Allow-Origin", "http://127.0.0.1", ); add_header( &mut response, "Access-Control-Allow-Methods", "GET, PUT, POST, DELETE, OPTIONS", ); add_header( &mut response, "Access-Control-Allow-Headers", "Content-Type, X-Tessera-Origin", ); response } /// 向响应追加经过 HTTP 库检查的头字段,不直接拼接未验证的原始报文。 fn add_header(response: &mut Response>>, name: &str, value: &str) { // 由 HTTP 库验证字段格式,非法头被忽略,不把未校验字符拼进原始响应报文。 if let Ok(header) = Header::from_bytes(name, value) { response.add_header(header); } } #[cfg(test)] mod tests;