test: cover deferred input results and one-shot event callbacks

src/api.rs

@@ -100,7 +100,9 @@ fn start_server_observed(
                             pending = Some((job, input));
                             thread::sleep(std::time::Duration::from_millis(1));
                         }
-                        RequestProgress::Complete(response) => { let _ = job.request.respond(response); }
+                        RequestProgress::Complete(response) => {
+                            let _ = job.request.respond(response);
+                        }
                     }
                     continue;
                 }
@@ -118,7 +120,9 @@ fn start_server_observed(
                     Ok(job) => {
                         match worker.begin_request(&job, &worker_namespace, &worker_notify) {
                             RequestProgress::Pending(input) => pending = Some((job, input)),
-                            RequestProgress::Complete(response) => { let _ = job.request.respond(response); }
+                            RequestProgress::Complete(response) => {
+                                let _ = job.request.respond(response);
+                            }
                         }
                     }
                     Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
@@ -447,7 +451,8 @@ fn submit(
 }
 
 type HttpResponse = Response<std::io::Cursor<Vec<u8>>>;
-type InputOutcome = std::thread::Result<Result<(endor_host::ReplResult, Vec<crate::window::Command>)>>;
+type InputOutcome =
+    std::thread::Result<Result<(endor_host::ReplResult, Vec<crate::window::Command>)>>;
 
 struct PendingInput {
     epoch: u64,
@@ -461,16 +466,20 @@ enum RequestProgress {
     Complete(HttpResponse),
 }
 
-fn current_viewports(current: &NamespaceState) -> std::collections::BTreeMap<u64, crate::layout::Size> {
+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
-                    }))
+                    && current
+                        .document
+                        .windows
+                        .get(id)
+                        .is_some_and(|view| view.frame.session == current.runtime_session))
         })
         .map(|(id, size)| (*id, *size))
         .collect();
@@ -677,7 +686,12 @@ impl Worker {
         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")),
+            Err(_) => {
+                return RequestProgress::Complete(error_response(
+                    StatusCode(500),
+                    "namespace lock is poisoned",
+                ));
+            }
         };
         if epoch != job.submission {
             return RequestProgress::Complete(superseded(job.submission));
@@ -703,7 +717,12 @@ impl Worker {
         let event = if job.kind == RequestKind::Event {
             let input: crate::events::Input = match serde_json::from_str(&job.source) {
                 Ok(input) => input,
-                Err(error) => return RequestProgress::Complete(error_response(StatusCode(400), error.to_string())),
+                Err(error) => {
+                    return RequestProgress::Complete(error_response(
+                        StatusCode(400),
+                        error.to_string(),
+                    ));
+                }
             };
             if !matches!(
                 input.kind.as_str(),
@@ -716,20 +735,31 @@ impl Worker {
                     | "tab-drop"
                     | "window-close"
             ) {
-                return RequestProgress::Complete(error_response(StatusCode(400), "unsupported event kind"));
+                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")),
+                Err(_) => {
+                    return RequestProgress::Complete(error_response(
+                        StatusCode(500),
+                        "namespace lock poisoned",
+                    ));
+                }
             };
             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"));
+                    return RequestProgress::Complete(error_response(
+                        StatusCode(409),
+                        "window is no longer active",
+                    ));
                 };
                 &document.frame
             };
@@ -748,7 +778,9 @@ impl Worker {
                         .is_some_and(|(_, runtime)| runtime.is_faulted());
                 if stopped {
                     drop(current);
-                    return RequestProgress::Complete(self.close_inactive_window(&input, epoch, namespace, notify));
+                    return RequestProgress::Complete(
+                        self.close_inactive_window(&input, epoch, namespace, notify),
+                    );
                 }
             }
             if let Some(target) = &input.tab_drop {
@@ -818,7 +850,12 @@ impl Worker {
         {
             self.next_session = match self.next_session.checked_add(1) {
                 Some(id) => id,
-                None => return RequestProgress::Complete(error_response(StatusCode(500), "session identities exhausted")),
+                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) {
@@ -830,12 +867,22 @@ impl Worker {
                     }
                     self.repl = Some((epoch, session));
                 }
-                Err(error) => return RequestProgress::Complete(error_response(StatusCode(500), error.to_string())),
+                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")),
+            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;
@@ -845,18 +892,36 @@ impl Worker {
             } else {
                 session.execute_incremental(&job.source, &job.namespace, &viewports)?
             };
-            let commands = if session.has_pending_refresh() { Vec::new() } else { session.take_window_commands()? };
+            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(Box::new(PendingInput { epoch, id, event: event.is_some(), result }));
+                return RequestProgress::Pending(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))
+        RequestProgress::Complete(self.finish_input(
+            job,
+            namespace,
+            notify,
+            epoch,
+            id,
+            event.is_some(),
+            faulted,
+            result,
+        ))
     }
 
     fn resume_request(
@@ -867,18 +932,28 @@ impl Worker {
         notify: &Notify,
     ) -> RequestProgress {
         let viewports = match namespace.read() {
-            Ok(current) if current.submission == input.epoch || current.runtime_session == input.id => current_viewports(&current),
+            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"));
+                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"));
+            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));
@@ -900,7 +975,16 @@ impl Worker {
             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))
+        RequestProgress::Complete(self.finish_input(
+            job,
+            namespace,
+            notify,
+            input.epoch,
+            input.id,
+            input.event,
+            faulted,
+            result,
+        ))
     }
 
     #[allow(clippy::too_many_arguments)]

src/endor_host.rs

@@ -993,7 +993,9 @@ impl Drop for ReplSession {
         {
             // Close the read-only continuation and checkpoint roots before
             // Candidate releases the globals/services of a discarded domain.
-            let _ = self.bridge.cancel_layout_frame(&mut self.candidate.vm, *pending);
+            let _ = self
+                .bridge
+                .cancel_layout_frame(&mut self.candidate.vm, *pending);
         }
     }
 }
@@ -4233,6 +4235,116 @@ page.render(block)
         assert_eq!(source_measure_stat(&mut session, "layoutCancels"), 1);
     }
 
+    #[test]
+    fn incremental_input_retains_value_and_never_replays_source_while_retargeting() {
+        let children = std::iter::repeat_n("item", 80)
+            .collect::<Vec<_>>()
+            .join(", ");
+        let source = format!(
+            "count = ui.state(0)\nitem = RectBlock(2,20)\npage = Page()\npage.render(Row(0,0,{children}))"
+        );
+        let (mut session, initial) = ReplSession::load(&source, "app.main", 275, 1).unwrap();
+        let first = std::collections::BTreeMap::new();
+        let result = session
+            .execute_incremental(
+                "count.increment()\nitem.width = 3\ncount.value",
+                "app.main",
+                &first,
+            )
+            .unwrap();
+        assert_eq!(result.value, "1");
+        assert!(result.document.is_none());
+        assert!(session.has_pending_refresh());
+        assert!(session.take_window_commands().is_err());
+        assert!(session.execute("count.increment()", "app.main").is_err());
+        for _ in 0..20 {
+            session.resize_windows_incremental(&first, false).unwrap();
+            if session.candidate.vm.has_suspended_call() {
+                break;
+            }
+        }
+        assert!(session.candidate.vm.has_suspended_call());
+        assert_eq!(
+            session.documents[&crate::window::MAIN_WINDOW]
+                .frame
+                .layout_version,
+            initial.frame.layout_version
+        );
+        let latest = std::collections::BTreeMap::from([(
+            crate::window::MAIN_WINDOW,
+            crate::layout::Size {
+                width: 900.0,
+                height: 650.0,
+            },
+        )]);
+        session.resize_windows_incremental(&latest, false).unwrap();
+        let settled = session.resize_windows(&latest).unwrap().unwrap();
+        assert_eq!(settled.frame.mounts.last().unwrap().x, 237.0);
+        assert_eq!(result.value, "1");
+        assert_eq!(
+            session.execute("count.value", "app.main").unwrap().value,
+            "1"
+        );
+        assert_eq!(source_measure_stat(&mut session, "layoutCancels"), 1);
+    }
+
+    #[test]
+    fn incremental_event_runs_callback_once_before_publishing_its_layout() {
+        let children = std::iter::repeat_n("item", 80)
+            .collect::<Vec<_>>()
+            .join(", ");
+        let source = format!(
+            "count = ui.state(0)\nitem = RectBlock(2,20)\nbutton = Button(\"increment\")\nbutton.onClick(count.increment)\npage = Page()\npage.render(Column(0,0,button,ui.textOf(count,\"value\"),Row(0,0,{children})))"
+        );
+        let (mut session, initial) = ReplSession::load(&source, "app.main", 276, 1).unwrap();
+        let mount = initial
+            .frame
+            .mounts
+            .iter()
+            .find(|mount| mount.interactive)
+            .unwrap();
+        let input = crate::events::Input {
+            window: crate::window::MAIN_WINDOW,
+            session: session.id,
+            layout_version: initial.frame.layout_version,
+            mount: mount.id,
+            kind: "click".into(),
+            point: None,
+            press_point: None,
+            delta: None,
+            tab_drop: None,
+            screen_position: None,
+        };
+        let viewports = std::collections::BTreeMap::new();
+        let result = session.dispatch_incremental(&input, &viewports).unwrap();
+        assert!(result.document.is_none());
+        assert!(session.has_pending_refresh());
+        assert_eq!(
+            session.documents[&crate::window::MAIN_WINDOW]
+                .frame
+                .layout_version,
+            initial.frame.layout_version
+        );
+        let mut complete = false;
+        for _ in 0..1000 {
+            if session
+                .resize_windows_incremental(&viewports, false)
+                .unwrap()
+                .is_some()
+            {
+                complete = true;
+                break;
+            }
+        }
+        assert!(complete);
+        assert!(!session.has_pending_refresh());
+        assert_eq!(
+            session.execute("count.value", "app.main").unwrap().value,
+            "1"
+        );
+        assert!(source_measure_stat(&mut session, "refreshYields") > 0);
+    }
+
     #[test]
     fn large_intrinsic_layout_yields_and_retargets_without_geometry_loss() {
         let children = std::iter::repeat_n("item", 80)