feat: execute Tessera UI source with embedded Aster

src/api.rs

@@ -1,6 +1,10 @@
 use std::{
     io::Read,
-    sync::{Arc, RwLock},
+    panic::{AssertUnwindSafe, catch_unwind},
+    sync::{
+        Arc, RwLock,
+        mpsc::{SyncSender, TrySendError, sync_channel},
+    },
     thread,
 };
 
@@ -8,14 +12,18 @@ use anyhow::{Context as _, Result};
 use tiny_http::{Header, Method, Request, Response, Server, StatusCode};
 use winit::event_loop::EventLoopProxy;
 
-use crate::{
-    app::AppEvent,
-    compiler,
-    document::{NamespaceState, RenderDocument},
-};
+use crate::{app::AppEvent, document::NamespaceState, endor_host};
 
 pub const API_ADDRESS: &str = "127.0.0.1:4737";
-const MAX_CODE_SIZE: u64 = 1_048_576;
+const MAX_CODE_SIZE: u64 = endor_host::MAX_SOURCE_BYTES as u64;
+type Notify = Arc<dyn Fn() + Send + Sync>;
+
+struct BuildRequest {
+    request: Request,
+    source: String,
+    namespace: String,
+    submission: u64,
+}
 
 pub fn start(
     namespace: Arc<RwLock<NamespaceState>>,
@@ -24,21 +32,45 @@ pub fn start(
     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);
+    });
+    start_server(Arc::new(server), namespace, notify)?;
+    Ok(())
+}
+
+fn start_server(
+    server: Arc<Server>,
+    namespace: Arc<RwLock<NamespaceState>>,
+    notify: Notify,
+) -> Result<thread::JoinHandle<()>> {
+    let (sender, receiver) = sync_channel::<BuildRequest>(8);
+    let worker_namespace = namespace.clone();
+    let worker_notify = notify.clone();
+    thread::Builder::new()
+        .name("tessera-endor-worker".into())
+        .spawn(move || {
+            for job in receiver {
+                let response = build(&job, &worker_namespace, &worker_notify);
+                let _ = job.request.respond(response);
+            }
+        })
+        .context("start Aster execution worker")?;
     thread::Builder::new()
         .name("tessera-agent-api".to_owned())
         .spawn(move || {
             for request in server.incoming_requests() {
-                handle(request, &namespace, &proxy);
+                handle(request, &namespace, &notify, &sender);
             }
         })
-        .context("start Tessera agent API thread")?;
-    Ok(())
+        .context("start Tessera agent API thread")
 }
 
 fn handle(
-    mut request: Request,
+    request: Request,
     namespace: &Arc<RwLock<NamespaceState>>,
-    proxy: &EventLoopProxy<AppEvent>,
+    notify: &Notify,
+    sender: &SyncSender<BuildRequest>,
 ) {
     let method = request.method().clone();
     let path = request
@@ -48,13 +80,19 @@ fn handle(
         .unwrap_or(request.url())
         .to_owned();
 
+    if method == Method::Put
+        && let Some(addressed) = namespace_from_path(&path)
+    {
+        submit(request, addressed, namespace, sender);
+        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(&mut request, method, &addressed_namespace, namespace, proxy)
+        code_request(method, &addressed_namespace, namespace, notify)
     } else if method == Method::Options {
         text(StatusCode(204), "", "text/plain")
     } else {
@@ -65,11 +103,10 @@ fn handle(
 }
 
 fn code_request(
-    request: &mut Request,
     method: Method,
     addressed_namespace: &str,
     namespace: &Arc<RwLock<NamespaceState>>,
-    proxy: &EventLoopProxy<AppEvent>,
+    notify: &Notify,
 ) -> Response<std::io::Cursor<Vec<u8>>> {
     let Ok(current) = namespace.read() else {
         return error_response(StatusCode(500), "namespace lock is poisoned");
@@ -100,43 +137,14 @@ fn code_request(
     drop(current);
 
     match method {
-        Method::Put => match read_code(request) {
-            Ok(code) => match compiler::compile(&code, addressed_namespace) {
-                Ok(document) => {
-                    if let Err(error) = document.validate() {
-                        return compile_error(error.to_string());
-                    }
-                    match namespace.write() {
-                        Ok(mut current) => {
-                            current.revision += 1;
-                            current.document = document;
-                            let revision = current.revision;
-                            drop(current);
-                            let _ = proxy.send_event(AppEvent::DocumentUpdated);
-                            json(
-                                StatusCode(200),
-                                serde_json::json!({
-                                    "ok": true,
-                                    "namespace": addressed_namespace,
-                                    "revision": revision,
-                                    "diagnostics": []
-                                }),
-                            )
-                        }
-                        Err(_) => error_response(StatusCode(500), "namespace lock is poisoned"),
-                    }
-                }
-                Err(error) => compile_error(error.to_string()),
-            },
-            Err(error) => error_response(StatusCode(400), error.to_string()),
-        },
         Method::Delete => match namespace.write() {
             Ok(mut current) => {
-                current.revision += 1;
-                current.document = RenderDocument::default();
-                let revision = current.revision;
+                let revision = match current.clear() {
+                    Ok(revision) => revision,
+                    Err(error) => return error_response(StatusCode(500), error.to_string()),
+                };
                 drop(current);
-                let _ = proxy.send_event(AppEvent::DocumentUpdated);
+                notify();
                 json(
                     StatusCode(200),
                     serde_json::json!({
@@ -153,6 +161,112 @@ fn code_request(
     }
 }
 
+fn submit(
+    mut request: Request,
+    addressed: &str,
+    namespace: &Arc<RwLock<NamespaceState>>,
+    sender: &SyncSender<BuildRequest>,
+) {
+    let source = match read_code(&mut request) {
+        Ok(source) => source,
+        Err(error) => {
+            let _ = request.respond(error_response(StatusCode(400), error.to_string()));
+            return;
+        }
+    };
+    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;
+    }
+    let submission = match current.next_submission() {
+        Ok(id) => id,
+        Err(error) => {
+            let _ = request.respond(error_response(StatusCode(500), error.to_string()));
+            return;
+        }
+    };
+    let job = BuildRequest {
+        request,
+        source,
+        namespace: addressed.into(),
+        submission,
+    };
+    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(()) => current.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 build(
+    job: &BuildRequest,
+    namespace: &Arc<RwLock<NamespaceState>>,
+    notify: &Notify,
+) -> Response<std::io::Cursor<Vec<u8>>> {
+    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",
+        ),
+    }
+}
+
+fn superseded(submission: u64) -> Response<std::io::Cursor<Vec<u8>>> {
+    json(
+        StatusCode(409),
+        serde_json::json!({ "ok": false, "submission": submission,
+        "error": "submission superseded by a newer PUT or DELETE" }),
+    )
+}
+
 fn namespace_from_path(path: &str) -> Option<&str> {
     path.strip_prefix("/v1/namespaces/")?
         .strip_suffix("/code")
@@ -180,13 +294,14 @@ fn read_code(request: &mut Request) -> Result<String> {
 
 fn agent_guide(namespace: &NamespaceState) -> String {
     format!(
-        r#"# Tessera Agent API
+        r##"# Tessera Agent API
 
 Current namespace: `{name}`
 Current revision: `{revision}`
 
-Tessera stores code by namespace. It has no files. Submit code only; Tessera parses,
-checks, compiles, and renders it.
+Submit ordinary Endor functions. Tessera embeds Aster's frontend, compiler and VM.
+The namespace comes from the request path. Declare a zero-argument `main` returning
+a UI node. Type annotations are currently syntax only in Aster.
 
 ## Code CRUD
 
@@ -196,42 +311,52 @@ checks, compiles, and renders it.
 
 Send PUT bodies as `text/plain; charset=utf-8`.
 
-## Minimal syntax
+```endor
+{example}```
 
-```tessera
-namespace {name}
+## UI host object
 
-use tessera.ui::{{Column, Row, Spacer, Text}}
+Positional arguments only:
 
-export scene Main {{
-    render {{
-        Column(gap: 12, padding: 24) {{
-            Text("Hello", role: .title)
-            Text("Rendered by Tessera")
-        }}
-    }}
-}}
-```
+- `ui.text(value, size = 1, color = "#242422")`; integer size 1..8.
+- `ui.rect(width, height, radius = 0, color = "#DCE8FF")`.
+- `ui.column(padding, gap, child...)` and `ui.row(padding, gap, child...)`.
+- `ui.spacer(size)`.
 
-Supported by this syntax experiment: one exported scene, one render root,
-`Column`, `Row`, `Spacer`, `Text`, `gap`, `padding`, and `.title`.
-Compilation errors reject the update and preserve the last rendered revision.
-"#,
+Defaults above describe the host API; do not write keyword arguments in Endor.
+Colors use #RRGGBB or #RRGGBBAA. Dimensions are finite numbers in 0..1000000.
+Limits: 1 MiB source, 4096 constructed nodes, 64 UI levels, 10000 expanded layout
+nodes, 16 KiB per text, 128 VM frames and 1000000 instructions per execution.
+
+PUT waits for its worker result. Errors return 422 with diagnostics and preserve
+the last successful revision. A newer accepted PUT or DELETE supersedes older
+work (409); a full eight-entry queue returns 503. DELETE immediately clears the
+document. GET remains available while Aster executes. The window redraws from an
+owned snapshot; resize does not rerun Endor. This first stage builds static UI;
+events, persistent VM state, Chinese font rendering and process isolation remain
+future work. It is not an untrusted-code sandbox.
+"##,
         name = namespace.name,
         revision = namespace.revision,
-        address = API_ADDRESS
+        address = API_ADDRESS,
+        example = include_str!("../examples/hello.endor"),
     )
 }
 
-fn compile_error(message: String) -> Response<std::io::Cursor<Vec<u8>>> {
+fn compile_error(error: anyhow::Error) -> Response<std::io::Cursor<Vec<u8>>> {
+    let mut diagnostic =
+        serde_json::json!({ "severity": "error", "message": format!("{error:#}") });
+    if let Some(source) = error.downcast_ref::<frontend::SourceDiagnostic>() {
+        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": [{
-                "severity": "error",
-                "message": message
-            }]
+            "diagnostics": [diagnostic]
         }),
     )
 }
@@ -285,6 +410,154 @@ fn add_header(response: &mut Response<std::io::Cursor<Vec<u8>>>, name: &str, val
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::{
+        io::Write,
+        net::{SocketAddr, TcpStream},
+        sync::atomic::{AtomicUsize, Ordering},
+        time::Duration,
+    };
+
+    struct TestServer {
+        server: Arc<Server>,
+        address: SocketAddr,
+        api: Option<thread::JoinHandle<()>>,
+    }
+
+    impl TestServer {
+        fn request(&self, method: &str, path: &str, body: &str) -> (u16, String, String) {
+            let mut stream = TcpStream::connect(self.address).unwrap();
+            stream
+                .set_read_timeout(Some(Duration::from_secs(10)))
+                .unwrap();
+            stream
+                .set_write_timeout(Some(Duration::from_secs(10)))
+                .unwrap();
+            write!(stream, "{method} {path} HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Length: {}\r\n\r\n{body}", body.len()).unwrap();
+            let mut response = String::new();
+            stream.read_to_string(&mut response).unwrap();
+            let (headers, body) = response.split_once("\r\n\r\n").unwrap();
+            let status = headers.split_whitespace().nth(1).unwrap().parse().unwrap();
+            (status, headers.to_owned(), body.to_owned())
+        }
+    }
+
+    impl Drop for TestServer {
+        fn drop(&mut self) {
+            self.server.unblock();
+            if let Some(api) = self.api.take() {
+                let _ = api.join();
+            }
+        }
+    }
+
+    #[test]
+    fn http_executes_real_endor_preserves_failed_updates_and_supports_delete() {
+        let server = Arc::new(Server::http("127.0.0.1:0").unwrap());
+        let namespace = Arc::new(RwLock::new(NamespaceState::default()));
+        let notifications = Arc::new(AtomicUsize::new(0));
+        let counter = notifications.clone();
+        let notify: Notify = Arc::new(move || {
+            counter.fetch_add(1, Ordering::SeqCst);
+        });
+        let api = start_server(server.clone(), namespace.clone(), notify).unwrap();
+        let client = TestServer {
+            address: server.server_addr().to_ip().unwrap(),
+            server,
+            api: Some(api),
+        };
+        let route = "/v1/namespaces/app.main/code";
+        let example = include_str!("../examples/hello.endor");
+        let (status, _, guide) = client.request("GET", "/agent", "");
+        assert_eq!(status, 200);
+        assert!(guide.contains("function main(): View"));
+        let (status, _, body) = client.request("PUT", route, example);
+        assert_eq!(status, 200, "{body}");
+        assert_eq!(
+            serde_json::from_str::<serde_json::Value>(&body).unwrap()["revision"],
+            1
+        );
+        let (status, headers, source) = client.request("GET", route, "");
+        assert_eq!(status, 200);
+        assert!(headers.to_lowercase().contains("x-tessera-revision: 1"));
+        assert_eq!(source, example);
+        let (status, _, body) =
+            client.request("PUT", route, "function main(): View { return missing }");
+        assert_eq!(status, 422, "{body}");
+        assert_eq!(
+            serde_json::from_str::<serde_json::Value>(&body).unwrap()["diagnostics"][0]["line"],
+            1
+        );
+        assert_eq!(client.request("GET", route, "").2, example);
+        let second = example.replace("Hello Endor", "Second");
+        assert_eq!(client.request("PUT", route, &second).0, 200);
+        assert_eq!(namespace.read().unwrap().revision, 2);
+        assert_eq!(client.request("DELETE", route, "").0, 200);
+        assert_eq!(client.request("GET", route, "").2, "");
+        assert!(namespace.read().unwrap().document.frame.commands.is_empty());
+        assert_eq!(client.request("PUT", route, example).0, 200);
+        assert_eq!(namespace.read().unwrap().revision, 4);
+        assert_eq!(
+            client
+                .request("PUT", "/v1/namespaces/other/code", example)
+                .0,
+            404
+        );
+        assert_eq!(notifications.load(Ordering::SeqCst), 4);
+    }
+
+    #[test]
+    fn http_delete_remains_available_and_invalidates_queued_put() {
+        use std::{
+            sync::{Mutex, mpsc},
+            time::Instant,
+        };
+        let server = Arc::new(Server::http("127.0.0.1:0").unwrap());
+        let namespace = Arc::new(RwLock::new(NamespaceState::default()));
+        let (entered_tx, entered_rx) = mpsc::channel();
+        let (release_tx, release_rx) = mpsc::channel();
+        let release_rx = Mutex::new(release_rx);
+        let calls = AtomicUsize::new(0);
+        let notify: Notify = Arc::new(move || {
+            if calls.fetch_add(1, Ordering::SeqCst) == 0 {
+                entered_tx.send(()).unwrap();
+                release_rx.lock().unwrap().recv().unwrap();
+            }
+        });
+        let api = start_server(server.clone(), namespace.clone(), notify).unwrap();
+        let client = TestServer {
+            address: server.server_addr().to_ip().unwrap(),
+            server,
+            api: Some(api),
+        };
+        // Unblock even if a test assertion fails, before scoped threads join.
+        struct Release(mpsc::Sender<()>);
+        impl Drop for Release {
+            fn drop(&mut self) {
+                let _ = self.0.send(());
+            }
+        }
+        thread::scope(|scope| {
+            let release = Release(release_tx);
+            let client = &client;
+            let route = "/v1/namespaces/app.main/code";
+            let example = include_str!("../examples/hello.endor");
+            let first = scope.spawn(move || client.request("PUT", route, example));
+            entered_rx.recv_timeout(Duration::from_secs(10)).unwrap();
+            let pending = scope.spawn(move || client.request("PUT", route, example));
+            let deadline = Instant::now() + Duration::from_secs(10);
+            while namespace.read().unwrap().submission < 2 {
+                assert!(Instant::now() < deadline, "second PUT was not admitted");
+                thread::sleep(Duration::from_millis(5));
+            }
+            assert_eq!(client.request("GET", route, "").2, example);
+            assert_eq!(client.request("DELETE", route, "").0, 200);
+            drop(release);
+            assert_eq!(first.join().unwrap().0, 200);
+            assert_eq!(pending.join().unwrap().0, 409);
+            assert_eq!(client.request("GET", route, "").2, "");
+            assert_eq!(namespace.read().unwrap().revision, 2);
+        });
+    }
 
     #[test]
     fn extracts_namespace_only_from_code_routes() {
@@ -300,6 +573,8 @@ mod tests {
     fn guide_contains_only_code_crud_routes() {
         let guide = agent_guide(&NamespaceState::default());
         assert!(guide.contains("Code CRUD"));
+        assert!(guide.contains("function main(): View"));
+        assert!(!guide.contains("export scene"));
         assert!(guide.contains("PUT http://127.0.0.1:4737"));
         assert!(!guide.contains("display_list"));
     }