W5-04b: Not-Aus in der App mit 10-Sekunden-Frist

src-tauri/src/estop.rs

@@ -0,0 +1,165 @@
+//! App side of the global emergency stop (W5-04b). The store half (W5-04a)
+//! owns the persistent barrier; this half ends the running agent processes
+//! within [`DEADLINE`] and says so only when it has observed that no session
+//! is left. Fail closed: a fleet that cannot be counted, or still has
+//! sessions at the deadline, is reported as not stopped - never as stopped.
+use std::time::{Duration, Instant};
+
+/// How long the app may take to end every running dispatch.
+pub const DEADLINE: Duration = Duration::from_secs(10);
+
+/// How often the fleet is re-checked (and re-signalled) while waiting.
+const POLL: Duration = Duration::from_millis(100);
+
+/// Time source, injectable so the deadline is tested without waiting.
+pub trait Clock {
+    fn now(&self) -> Instant;
+    fn sleep(&self, duration: Duration);
+}
+
+pub struct SystemClock;
+
+impl Clock for SystemClock {
+    fn now(&self) -> Instant {
+        Instant::now()
+    }
+    fn sleep(&self, duration: Duration) {
+        std::thread::sleep(duration);
+    }
+}
+
+/// The running agent processes, as far as the stop can reach them.
+pub trait Fleet {
+    /// Signal every session, including reservations not yet spawned.
+    fn kill_all(&self);
+    /// Sessions still alive; an error means "cannot tell".
+    fn live(&self) -> Result<usize, String>;
+}
+
+impl Fleet for crate::pty::PtyManager {
+    fn kill_all(&self) {
+        crate::pty::PtyManager::kill_all(self);
+    }
+    fn live(&self) -> Result<usize, String> {
+        self.live_session_ids().map(|ids| ids.len())
+    }
+}
+
+/// Kill the fleet and wait until it is observed empty. Signals again on every
+/// poll so a spawn that raced the first signal is caught too.
+pub fn enforce(fleet: &dyn Fleet, clock: &dyn Clock, deadline: Duration) -> Result<(), String> {
+    let end = clock.now() + deadline;
+    loop {
+        fleet.kill_all();
+        let remaining = fleet.live().map_err(|error| {
+            format!("emergency stop: cannot confirm that agents ended: {error}")
+        })?;
+        if remaining == 0 {
+            return Ok(());
+        }
+        if clock.now() >= end {
+            return Err(format!(
+                "emergency stop: {remaining} agent session(s) still alive after {}s",
+                deadline.as_secs()
+            ));
+        }
+        clock.sleep(POLL);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::cell::Cell;
+
+    /// Virtual clock: `sleep` advances time and never waits.
+    struct FakeClock {
+        start: Instant,
+        elapsed: Cell<Duration>,
+    }
+
+    impl FakeClock {
+        fn new() -> Self {
+            Self {
+                start: Instant::now(),
+                elapsed: Cell::new(Duration::ZERO),
+            }
+        }
+    }
+
+    impl Clock for FakeClock {
+        fn now(&self) -> Instant {
+            self.start + self.elapsed.get()
+        }
+        fn sleep(&self, duration: Duration) {
+            self.elapsed.set(self.elapsed.get() + duration);
+        }
+    }
+
+    /// Dies `dies_after` into the fake timeline; `None` never dies.
+    struct FakeFleet<'a> {
+        clock: &'a FakeClock,
+        dies_after: Option<Duration>,
+        signals: Cell<u32>,
+        broken: bool,
+    }
+
+    impl Fleet for FakeFleet<'_> {
+        fn kill_all(&self) {
+            self.signals.set(self.signals.get() + 1);
+        }
+        fn live(&self) -> Result<usize, String> {
+            if self.broken {
+                return Err("registry poisoned".into());
+            }
+            Ok(match self.dies_after {
+                Some(after) if self.clock.elapsed.get() >= after => 0,
+                _ => 1,
+            })
+        }
+    }
+
+    fn fleet(clock: &FakeClock, dies_after: Option<Duration>, broken: bool) -> FakeFleet<'_> {
+        FakeFleet {
+            clock,
+            dies_after,
+            signals: Cell::new(0),
+            broken,
+        }
+    }
+
+    #[test]
+    fn a_fleet_that_ends_inside_the_deadline_is_stopped() {
+        let clock = FakeClock::new();
+        let fleet = fleet(&clock, Some(Duration::from_secs(9)), false);
+        enforce(&fleet, &clock, DEADLINE).unwrap();
+        assert!(clock.elapsed.get() <= DEADLINE);
+        assert!(fleet.signals.get() > 1, "survivors are signalled again");
+    }
+
+    #[test]
+    fn a_fleet_alive_at_the_deadline_is_reported_not_stopped() {
+        let clock = FakeClock::new();
+        let fleet = fleet(&clock, None, false);
+        let error = enforce(&fleet, &clock, DEADLINE).unwrap_err();
+        assert!(error.contains("still alive after 10s"), "{error}");
+        assert!(clock.elapsed.get() >= DEADLINE);
+        assert!(clock.elapsed.get() <= DEADLINE + POLL, "no overshoot");
+    }
+
+    #[test]
+    fn a_fleet_that_cannot_be_counted_is_never_reported_stopped() {
+        let clock = FakeClock::new();
+        let fleet = fleet(&clock, Some(Duration::ZERO), true);
+        let error = enforce(&fleet, &clock, DEADLINE).unwrap_err();
+        assert!(error.contains("cannot confirm"), "{error}");
+    }
+
+    #[test]
+    fn an_empty_fleet_returns_without_waiting() {
+        let clock = FakeClock::new();
+        let fleet = fleet(&clock, Some(Duration::ZERO), false);
+        enforce(&fleet, &clock, DEADLINE).unwrap();
+        assert_eq!(clock.elapsed.get(), Duration::ZERO);
+    }
+}

src-tauri/src/main.rs

@@ -72,6 +72,7 @@ mod diagnosis;
 mod diff;
 mod digest;
 mod enhance;
+mod estop;
 mod freetier;
 mod gh;
 mod hooks;
@@ -1132,6 +1133,42 @@ fn set_updater_state(
     Ok(())
 }
 
+// -- emergency stop (W5-04b) ----------------------------------------------
+
+/// The persistent barrier. An unreadable state is an error, which the UI
+/// shows as "stopped" - never as permission to dispatch.
+#[tauri::command]
+async fn get_emergency_stop(store: State<'_, Store>) -> Result<bool, String> {
+    store.emergency_stop_active().await
+}
+
+/// Raise or clear the global emergency stop. Raising writes the barrier first
+/// (new dispatches are refused from that moment, `set_emergency_stop` also
+/// revokes claims in flight), then ends every agent session within
+/// `estop::DEADLINE`; an `Err` means the end was not observed and the barrier
+/// stays up. Clearing only lifts the barrier - nothing is revived.
+#[tauri::command]
+async fn set_emergency_stop(
+    store: State<'_, Store>,
+    manager: State<'_, PtyManager>,
+    gate: State<'_, tokio::sync::Mutex<()>>,
+    active: bool,
+) -> Result<(), String> {
+    let _guard = gate.lock().await;
+    store.set_emergency_stop(active, "app").await?;
+    if !active {
+        manager.set_emergency_stop_active(false);
+        return Ok(());
+    }
+    manager.set_emergency_stop_active(true);
+    let manager = manager.inner().clone();
+    tauri::async_runtime::spawn_blocking(move || {
+        estop::enforce(&manager, &estop::SystemClock, estop::DEADLINE)
+    })
+    .await
+    .map_err(|error| format!("emergency stop: {error}"))?
+}
+
 // -- task queue (Phase 7) -------------------------------------------------
 
 #[tauri::command]
@@ -3440,6 +3477,7 @@ fn main() {
         .plugin(tauri_plugin_opener::init())
         .plugin(tauri_plugin_updater::Builder::new().build())
         .manage(PtyManager::default())
+        .manage(tokio::sync::Mutex::new(()))
         // The web interface starts on demand; its state only exists so the
         // frontend can ask for status or stop it later.
         .manage(Mutex::new(WebInterfaceState::default()))
@@ -3481,6 +3519,14 @@ fn main() {
             // would fail, which is why it is the first thing done with `dir`.
             learnings::set_data_dir(&dir);
             let store = init_store(&handle, &dir)?;
+            let stopped = tauri::async_runtime::block_on(store.emergency_stop_active())
+                .unwrap_or_else(|error| {
+                    eprintln!("projecta: emergency stop state unavailable at startup: {error}");
+                    true
+                });
+            handle
+                .state::<PtyManager>()
+                .set_emergency_stop_active(stopped);
             // W2-06: committed supervisor changes reach the window as runtime
             // notifications; the payload holds ids and reason codes only.
             let notice_app = handle.clone();
@@ -3671,6 +3717,8 @@ fn main() {
             write_pty,
             resize_pty,
             kill_pty,
+            get_emergency_stop,
+            set_emergency_stop,
             get_scrollback,
             get_session_restore,
             list_agent_profiles,
@@ -4390,6 +4438,10 @@ mod tests {
                 ".manage(Mutex::new(WebInterfaceState::default()))",
             ),
             ("PtyManager", ".manage(PtyManager::default())"),
+            (
+                "tokio::sync::Mutex<()>",
+                ".manage(tokio::sync::Mutex::new(()))",
+            ),
             ("Store", "app.manage(store)"),
             ("PanicNotice", "app.manage(panic_notice)"),
         ];
@@ -4449,4 +4501,20 @@ mod tests {
             );
         }
     }
+
+    #[test]
+    fn emergency_stop_changes_are_serialized() {
+        const SOURCE: &str = include_str!("main.rs");
+        let command = SOURCE
+            .split("async fn set_emergency_stop")
+            .nth(1)
+            .expect("set command")
+            .split("// -- task queue")
+            .next()
+            .expect("command boundary");
+        assert!(
+            command.contains("gate.lock().await"),
+            "raising and clearing the stop must not overlap"
+        );
+    }
 }

src-tauri/src/pty.rs

@@ -572,6 +572,7 @@ impl Drop for StartingSession {
 pub struct PtyManager {
     sessions: Arc<Mutex<HashMap<String, SessionEntry>>>,
     installation_started: Arc<AtomicBool>,
+    emergency_stop_active: Arc<AtomicBool>,
     next_id: Arc<AtomicU64>,
     on_exit: Arc<Mutex<Option<ExitHook>>>,
     on_output: Arc<Mutex<Option<OutputHook>>>,
@@ -582,6 +583,7 @@ impl Default for PtyManager {
         Self {
             sessions: Arc::new(Mutex::new(HashMap::new())),
             installation_started: Arc::new(AtomicBool::new(false)),
+            emergency_stop_active: Arc::new(AtomicBool::new(false)),
             next_id: Arc::new(AtomicU64::new(1)),
             on_exit: Arc::new(Mutex::new(None)),
             on_output: Arc::new(Mutex::new(None)),
@@ -734,10 +736,19 @@ impl PtyManager {
         if self.installation_started.load(Ordering::Relaxed) {
             return Err("Installation has started; restart before starting sessions".into());
         }
+        if self.emergency_stop_active.load(Ordering::Acquire) {
+            return Err(
+                "Global emergency stop is active; clear it before starting sessions".into(),
+            );
+        }
         registry.insert(session_id.clone(), SessionEntry::Reserved);
         Ok(session_id)
     }
 
+    pub fn set_emergency_stop_active(&self, active: bool) {
+        self.emergency_stop_active.store(active, Ordering::Release);
+    }
+
     /// Cancel only a not-yet-consumed reservation; never touches a live child.
     ///
     /// Same recovery as [`StartingSession::drop`] and the same reason: a
@@ -1230,6 +1241,7 @@ impl PtyManager {
 
     /// Kill every session, so app shutdown never orphans an agent process.
     pub fn kill_all(&self) {
+        self.set_emergency_stop_active(true);
         // Shutdown recovers a poisoned registry (W1-15b): returning here
         // left every agent process running after the app was gone. The loop
         // only flags and collects, so a recovered map is safe to walk.
@@ -2059,6 +2071,16 @@ mod tests {
         assert!(manager.reserve_session().is_err());
     }
 
+    #[test]
+    fn kill_all_blocks_new_sessions_until_the_stop_is_cleared() {
+        let manager = PtyManager::default();
+        manager.kill_all();
+        assert!(manager.reserve_session().is_err());
+        manager.set_emergency_stop_active(false);
+        let id = manager.reserve_session().unwrap();
+        manager.cancel_reservation(&id);
+    }
+
     /// Shutdown under poison used to return before touching a single session,
     /// leaving every agent process running after the app was gone. A
     /// reservation is the one entry a test can check without a real child:

src/components/EmergencyStop.test.tsx

@@ -0,0 +1,43 @@
+import { render, screen, fireEvent, waitFor } from "@testing-library/react";
+import { describe, it, expect, vi, beforeEach } from "vitest";
+
+import EmergencyStop from "./EmergencyStop";
+
+const ipc = vi.hoisted(() => ({
+  getEmergencyStop: vi.fn(),
+  setEmergencyStop: vi.fn(),
+}));
+vi.mock("../lib/ipc", () => ({ ...ipc, describeError: (e: unknown) => String(e) }));
+
+describe("EmergencyStop", () => {
+  beforeEach(() => {
+    ipc.getEmergencyStop.mockReset();
+    ipc.setEmergencyStop.mockReset();
+  });
+
+  it("raises the stop and shows it active", async () => {
+    ipc.getEmergencyStop.mockResolvedValue(false);
+    ipc.setEmergencyStop.mockResolvedValue(undefined);
+    render(<EmergencyStop />);
+    fireEvent.click(await screen.findByRole("button", { name: "Not-Aus auslösen" }));
+    expect(await screen.findByRole("button", { name: "Not-Aus aufheben" })).toBeTruthy();
+    expect(ipc.setEmergencyStop).toHaveBeenCalledWith(true);
+  });
+
+  it("treats an unreadable state as stopped", async () => {
+    ipc.getEmergencyStop.mockRejectedValue("db down");
+    render(<EmergencyStop />);
+    expect(await screen.findByRole("button", { name: "Not-Aus aufheben" })).toBeTruthy();
+  });
+
+  it("keeps the stop up and shows the error when the end was not confirmed", async () => {
+    ipc.getEmergencyStop.mockResolvedValue(false);
+    ipc.setEmergencyStop.mockRejectedValue("1 agent session(s) still alive after 10s");
+    render(<EmergencyStop />);
+    fireEvent.click(await screen.findByRole("button", { name: "Not-Aus auslösen" }));
+    await waitFor(() => expect(screen.getByRole("alert")).toHaveTextContent(/still alive after 10s/));
+    expect(screen.getByRole("button", { name: "Not-Aus aufheben" })).toBeTruthy();
+    expect(screen.getByRole("status")).toHaveTextContent(/Stillstand ist nicht bestätigt/);
+    expect(screen.queryByText(/alle Agenten sind beendet/)).toBeNull();
+  });
+});