lorenzo/start kw integration pr 6

Cargo.toml

@@ -13,7 +13,7 @@ mockall = "0.13.1"
 derive-getters = { version = "0.5.0", features = ["auto_copy_getters"] }
 lazy_static = "1.5.0"
 patch-hub-proc-macros = "0.1.0"
-ratatui = "0.29.0"
+ratatui = { version = "0.29.0", features = ["unstable-rendered-line-info"] }
 regex = "1.11.1"
 serde = { version = "1.0.219", features = ["derive"] }
 serde-xml-rs = "0.8.1"

src/app/actor.rs

@@ -3,30 +3,41 @@
 //! Each frame: process system updates → project state to [`AppViewModel`] via
 //! [`UiHandle`](crate::ui::handle::UiHandle) → draw through
 //! [`TerminalHandle`](crate::terminal::handle::TerminalHandle) → await the next
-//! [`InputEvent`](crate::input::event::InputEvent) from the channel registered
-//! with [`InputHandle`](crate::input::handle::InputHandle).
+//! [`InputEvent`](crate::input::event::InputEvent), a kw-status change, or a
+//! KwOps log-tail tick while a job is running on that screen.
 //!
 //! The actor stops when the input event channel closes (user quit) or when I/O
 //! returns an unrecoverable error. Startup dependency checks run before this
 //! actor is spawned.
-use std::ops::ControlFlow;
+use std::{ops::ControlFlow, time::Duration};
 
 use color_eyre::{eyre::eyre, Result};
-use tokio::{spawn, sync::mpsc};
+use tokio::{spawn, sync::mpsc, sync::watch, time::MissedTickBehavior};
 
 use crate::{
     app::{
         flows::{
-            bookmarked::handle_bookmarked_patchsets, details_actions::handle_patchset_details,
-            edit_config::handle_edit_config, latest::handle_latest_patchsets,
+            bookmarked::handle_bookmarked_patchsets,
+            details_actions::handle_patchset_details,
+            edit_config::handle_edit_config,
+            kw_ops::{
+                apply_kw_snapshot, fallback_kw_status, handle_kw_ops, poll_kw_status,
+                refresh_kw_ops_log_tail,
+            },
+            latest::handle_latest_patchsets,
             mail_list::handle_mailing_list_selection,
         },
         handle::AppHandle,
         loading::{terminal_error, TerminalLoadingIndicator},
+        popup::{AppPopup, ConfirmAction},
         screens::CurrentScreen,
         App,
     },
     input::{event::InputEvent, handle::InputHandle},
+    kw::{
+        errors::KwError,
+        status::{KwJobStatus, KwStatusSnapshot},
+    },
     terminal::{handle::TerminalHandle, messages::TerminalFrame},
     ui::handle::UiHandle,
 };
@@ -46,6 +57,13 @@ pub struct AppActor {
     event_rx: mpsc::Receiver<InputEvent>,
 }
 
+enum KwWatchEvent {
+    Updated(KwStatusSnapshot),
+    Closed,
+}
+
+const KW_OPS_LOG_TICK: Duration = Duration::from_millis(350);
+
 impl AppActor {
     /// Moves all resources into a new `AppActor`, spawns it on the Tokio
     /// runtime, and returns an [`AppHandle`] to wait on it.
@@ -71,45 +89,151 @@ impl AppActor {
         tracing::info!("app actor started");
         tracing::info!("app actor initialized");
 
+        let mut kw_status_rx = self.subscribe_kw_status().await;
         let mut loading = TerminalLoadingIndicator::new(self.terminal_handle.clone());
+        let mut log_interval = tokio::time::interval_at(
+            tokio::time::Instant::now() + KW_OPS_LOG_TICK,
+            KW_OPS_LOG_TICK,
+        );
+        log_interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
 
+        let mut redraw = true;
         loop {
-            self.app.process_system_updates(&mut loading).await?;
-
-            let scene = self
-                .ui_handle
-                .build_scene(self.app.present())
-                .await
-                .map_err(|e| eyre!("{e}"))?;
-            self.terminal_handle
-                .draw(TerminalFrame::Main(Box::new(scene)))
-                .await
-                .map_err(terminal_error)?;
-
-            match self.event_rx.recv().await {
-                Some(event) => {
-                    match on_input(&mut self.app, event, &self.terminal_handle, &mut loading)
-                        .await?
-                    {
-                        ControlFlow::Continue(()) => {
-                            self.input_handle
-                                .update_context(self.app.input_context())
-                                .await
-                                .ok();
+            let system_changed = self.app.process_system_updates(&mut loading).await?;
+            if system_changed {
+                redraw = true;
+            }
+
+            if redraw {
+                let scene = self
+                    .ui_handle
+                    .build_scene(self.app.present())
+                    .await
+                    .map_err(|e| eyre!("{e}"))?;
+                self.terminal_handle
+                    .draw(TerminalFrame::Main(Box::new(scene)))
+                    .await
+                    .map_err(terminal_error)?;
+            }
+
+            let tail_while_running = kw_ops_should_tail(&self.app);
+            let poll_status = kw_status_rx.is_none() && should_poll_kw_status(&self.app);
+            tokio::select! {
+                event = self.event_rx.recv() => {
+                    match event {
+                        Some(event) => {
+                            match on_input(&mut self.app, event, &self.terminal_handle, &mut loading)
+                                .await?
+                            {
+                                ControlFlow::Continue(()) => {
+                                    self.input_handle
+                                        .update_context(self.app.input_context())
+                                        .await
+                                        .ok();
+                                    redraw = true;
+                                }
+                                ControlFlow::Break(()) => break,
+                            }
+                        }
+                        None => {
+                            tracing::info!("input channel closed; app actor stopping");
+                            break;
                         }
-                        ControlFlow::Break(()) => break,
                     }
                 }
-                None => {
-                    tracing::info!("input channel closed; app actor stopping");
-                    break;
+                watch_event = kw_status_changed(&mut kw_status_rx) => {
+                    match watch_event {
+                        KwWatchEvent::Updated(snapshot) => {
+                            apply_kw_snapshot(&mut self.app, snapshot);
+                            if self.app.state.navigation.current_screen == CurrentScreen::KwOps
+                            {
+                                refresh_kw_ops_log_tail(&mut self.app).await;
+                            }
+                        }
+                        KwWatchEvent::Closed => {
+                            tracing::warn!(
+                                "kw status watch closed; falling back to keyboard-only redraws"
+                            );
+                            kw_status_rx = None;
+                            fallback_kw_status(&mut self.app).await;
+                        }
+                    }
+                    redraw = true;
+                }
+                _ = log_interval.tick(), if tail_while_running || poll_status => {
+                    let mut changed = false;
+                    if poll_status {
+                        changed |= poll_kw_status(&mut self.app).await;
+                    }
+                    if tail_while_running {
+                        changed |= refresh_kw_ops_log_tail(&mut self.app).await;
+                    }
+                    redraw = changed;
                 }
             }
         }
 
         tracing::info!("app actor stopped");
         Ok(())
     }
+
+    /// Subscribes once at startup. A missing handle or a failed
+    /// subscription leaves the app on keyboard-only redraws rather than
+    /// failing to start. A failed subscribe still seeds the projection
+    /// from GetStatus so a job already running is visible and the poll
+    /// arm can engage.
+    async fn subscribe_kw_status(&mut self) -> Option<watch::Receiver<KwStatusSnapshot>> {
+        let kw = self.app.services.kw.clone()?;
+        match kw.watch_status().await {
+            Ok(mut rx) => {
+                apply_kw_snapshot(&mut self.app, rx.borrow_and_update().clone());
+                Some(rx)
+            }
+            Err(error) => {
+                tracing::warn!(
+                    %error,
+                    "failed to subscribe to kw status; keyboard-only redraws"
+                );
+                fallback_kw_status(&mut self.app).await;
+                None
+            }
+        }
+    }
+}
+
+/// Waits for the next kw-status change. With no receiver this future
+/// never completes, so `select!` stays on the input arm instead of
+/// spinning.
+async fn kw_status_changed(rx: &mut Option<watch::Receiver<KwStatusSnapshot>>) -> KwWatchEvent {
+    match rx.as_mut() {
+        Some(rx) => match rx.changed().await {
+            Ok(()) => KwWatchEvent::Updated(rx.borrow_and_update().clone()),
+            Err(_) => KwWatchEvent::Closed,
+        },
+        None => std::future::pending().await,
+    }
+}
+
+fn kw_ops_should_tail(app: &App) -> bool {
+    app.state.navigation.current_screen == CurrentScreen::KwOps
+        && matches!(
+            app.state.kw.status.as_ref().map(|status| &status.job),
+            Some(KwJobStatus::Running { .. })
+        )
+}
+
+/// When the status watch is missing, poll GetStatus while a start is
+/// in flight or a job is running so `start_requested` cannot latch and
+/// the nav indicator can leave "building".
+fn should_poll_kw_status(app: &App) -> bool {
+    app.services.kw.is_some()
+        && (kw_job_is_running(app)
+            || app
+                .state
+                .kw
+                .ops
+                .as_ref()
+                .is_some_and(|ops| ops.start_requested))
 }
 
 async fn on_input(
@@ -119,11 +243,24 @@ async fn on_input(
     loading: &mut TerminalLoadingIndicator,
 ) -> Result<ControlFlow<()>> {
     if let Some(popup) = app.state.popup.as_mut() {
-        if input == InputEvent::ClosePopup {
-            app.state.popup = None;
-        } else {
-            popup.handle_scroll(input);
+        match input {
+            InputEvent::ClosePopup => {
+                app.state.popup = None;
+            }
+            InputEvent::ConfirmPopup => match popup.selected_confirm_action() {
+                Some(ConfirmAction::CancelKwAndQuit) => {
+                    app.state.popup = None;
+                    return Ok(cancel_kw_and_quit(app).await);
+                }
+                Some(ConfirmAction::Wait) => {
+                    app.state.popup = None;
+                }
+                None => {}
+            },
+            _ => popup.handle_input(input),
         }
+    } else if input == InputEvent::Quit && kw_job_is_running(app) {
+        app.state.popup = Some(AppPopup::quit_while_job_running());
     } else {
         tracing::debug!(screen = ?app.state.navigation.current_screen, "dispatching input to screen handler");
         match app.state.navigation.current_screen {
@@ -145,11 +282,38 @@ async fn on_input(
             CurrentScreen::LatestPatchsets => {
                 handle_latest_patchsets(app, input, loading).await?;
             }
+            CurrentScreen::KwOps => {
+                handle_kw_ops(app, input).await?;
+            }
         }
     }
     Ok(ControlFlow::Continue(()))
 }
 
+fn kw_job_is_running(app: &App) -> bool {
+    matches!(
+        app.state.kw.status.as_ref().map(|status| &status.job),
+        Some(KwJobStatus::Running { .. })
+    )
+}
+
+/// Cancel then leave. A job that finished while the confirm popup was
+/// open is `NoJobRunning`; that race is harmless and still quits.
+async fn cancel_kw_and_quit(app: &App) -> ControlFlow<()> {
+    if let Some(kw) = app.services.kw.as_ref() {
+        match kw.cancel().await {
+            Ok(()) => {}
+            Err(KwError::NoJobRunning) => {
+                tracing::debug!("kw job already finished before cancel-and-quit");
+            }
+            Err(error) => {
+                tracing::warn!(%error, "kw cancel failed while quitting");
+            }
+        }
+    }
+    ControlFlow::Break(())
+}
+
 #[cfg(test)]
 mod tests {
     use std::{collections::HashMap, sync::Arc};
@@ -226,12 +390,13 @@ mod tests {
                 config_state: ConfigUiState { edit_config: None },
                 config: ConfigState::default().to_snapshot(),
                 popup: None,
+                kw: Default::default(),
             },
             services: AppServices {
                 lore_api: LoreApiHandle::new(lore_tx),
                 render: RenderHandle::new(render_tx),
                 shell: Box::new(MockShellTrait::new()),
-                fs: Box::new(MockFileSystemTrait::new()),
+                fs: Arc::new(MockFileSystemTrait::new()),
                 config: dummy_config_handle(),
                 kw_history: Arc::new(MockKwHistoryStore::new()),
                 kw: None,
@@ -332,7 +497,7 @@ mod tests {
             ConfigState::default().to_snapshot(),
             dummy_config_handle(),
             bootstrap,
-            Box::new(MockFileSystemTrait::new()),
+            Arc::new(MockFileSystemTrait::new()),
             Box::new(MockShellTrait::new()),
             lore_api.clone(),
             render.clone(),

src/app/flows/details_actions.rs

@@ -17,6 +17,10 @@ pub async fn handle_patchset_details(
     input: InputEvent,
     terminal_handle: &TerminalHandle,
 ) -> Result<()> {
+    if input == InputEvent::OpenKwOps {
+        return crate::app::flows::kw_ops::open_kw_ops(app).await;
+    }
+
     let patchset_details_and_actions = app
         .state
         .lore
@@ -139,5 +143,6 @@ pub fn generate_help_popup() -> AppPopup {
         .keybind("r", "Toggle reply with Reviewed-by action")
         .keybind("Shift+r", "Toggle reply with Reviewed-by action for all patches")
         .keybind("Ctrl+t", "Show code-review trailers details")
+        .keybind("w", "Kw operations")
         .build()
 }