host: add supervisor-gated clean core.restart RPC

Return success before scheduling the existing shutdown path, coalesce pending
requests, and mark #loop invocations with SAUSAGE_LOOP=1. Keep core.exit
compatibility and document explicit per-call approval for all agents.

Card: SAU-134

cli/src/core.rs

@@ -62,6 +62,7 @@ impl Host {
             }
             "log" => Ok(self.0.journal.query(p).await),
             "activate" | "revert" => self.activation(caller, method, p).await,
+            "restart" => self.request_restart(),
             "exit" => {
                 self.0.exit.send_replace(true);
                 Ok(json!({}))

cli/src/host.rs

@@ -26,6 +26,8 @@ pub(crate) struct Inner {
     pub(crate) journal: Arc<journal::Journal>,
     pub(crate) notifications: tokio::sync::broadcast::Sender<Value>,
     pub(crate) exit: watch::Sender<bool>,
+    pub(crate) loop_supervised: bool,
+    pub(crate) restart_pending: std::sync::atomic::AtomicBool,
     pub(crate) components: components::Components,
     /// Calls hold it for reading from the choice of a version until their
     /// process is in the table. Activation holds it for writing while it
@@ -98,6 +100,8 @@ impl Host {
             }),
             notifications: tokio::sync::broadcast::channel(1024).0,
             exit: watch::channel(false).0,
+            loop_supervised: crate::restart::enabled(std::env::var_os("SAUSAGE_LOOP").as_deref()),
+            restart_pending: std::sync::atomic::AtomicBool::new(false),
             components: components::Components::default(),
             admission: tokio::sync::RwLock::new(()),
             shutdown: std::sync::atomic::AtomicBool::new(false),
@@ -143,12 +147,12 @@ impl Host {
         )
         .await
     }
-    /// Wait until a core.exit handler requests a restart.
+    /// Wait until core.restart (after its response grace) or core.exit requests shutdown.
     pub async fn exit_requested(&self) {
         let mut exit = self.0.exit.subscribe();
         let _ = exit.wait_for(|requested| *requested).await;
     }
-    /// Return whether a core.exit handler requested a restart.
+    /// Return whether a restart has reached the normal shutdown path.
     pub fn restart_requested(&self) -> bool {
         *self.0.exit.borrow()
     }

cli/src/lib.rs

@@ -12,6 +12,7 @@ mod journal;
 mod locks;
 mod native;
 mod process;
+mod restart;
 mod routing;
 pub mod runtime;
 mod services;

cli/src/restart.rs

@@ -0,0 +1,119 @@
+//! Supervisor-gated restart requests; the executable still owns normal shutdown.
+use crate::{Host, RpcResult, error};
+use serde_json::json;
+use std::{ffi::OsStr, sync::atomic::Ordering, time::Duration};
+
+pub(crate) fn enabled(value: Option<&OsStr>) -> bool {
+    value == Some(OsStr::new("1"))
+}
+
+impl Host {
+    pub(crate) fn request_restart(&self) -> RpcResult {
+        if !self.0.loop_supervised {
+            return Err(error(
+                -32600,
+                "restart needs the loop supervisor (SAUSAGE_LOOP not set); start with SAUSAGE_LOOP=1",
+            ));
+        }
+        if !self.0.restart_pending.swap(true, Ordering::SeqCst) {
+            let host = self.clone();
+            tokio::spawn(async move {
+                // Let the RPC response reach the caller before closing any stdin,
+                // including that of a detached gateway carrying this response.
+                tokio::time::sleep(Duration::from_millis(250)).await;
+                host.0.exit.send_replace(true);
+            });
+        }
+        Ok(json!({}))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{runtime::Runtime, world::World};
+    use std::sync::Arc;
+    use tokio::time::timeout;
+
+    #[test]
+    fn only_explicit_loop_marker_enables_restart() {
+        assert!(!enabled(None));
+        for value in ["", "0", "true", "yes", "2"] {
+            assert!(!enabled(Some(OsStr::new(value))));
+        }
+        assert!(enabled(Some(OsStr::new("1"))));
+    }
+
+    async fn host(supervised: bool) -> (tempfile::TempDir, Host) {
+        let dir = tempfile::tempdir().unwrap();
+        World::create(&dir.path().join("world")).unwrap();
+        let runtime = Runtime::new(&dir.path().join("cache")).unwrap();
+        let mut host = Host::test_world(&dir.path().join("world"), runtime)
+            .await
+            .unwrap();
+        // No global environment mutation: parallel tests have independent gates.
+        Arc::get_mut(&mut host.0).unwrap().loop_supervised = supervised;
+        (dir, host)
+    }
+
+    #[tokio::test]
+    async fn restart_without_supervisor_never_triggers_exit() {
+        timeout(Duration::from_secs(10), async {
+            let (_dir, host) = host(false).await;
+            let error = host.call("core.restart", json!({})).await.unwrap_err();
+            assert_eq!(error["code"], -32600);
+            assert!(
+                error["message"]
+                    .as_str()
+                    .unwrap()
+                    .contains("SAUSAGE_LOOP not set")
+            );
+            assert!(!host.0.restart_pending.load(Ordering::SeqCst));
+            assert!(
+                timeout(Duration::from_millis(300), host.exit_requested())
+                    .await
+                    .is_err()
+            );
+            host.stop().await;
+        })
+        .await
+        .unwrap();
+    }
+
+    #[tokio::test]
+    async fn restart_answers_before_trigger_and_duplicates_are_noops() {
+        timeout(Duration::from_secs(10), async {
+            let (_dir, host) = host(true).await;
+            // Same dispatch as a plugin's core.call, without killing this runner.
+            assert_eq!(
+                host.call("core.call", json!({"method":"core.restart"}))
+                    .await
+                    .unwrap(),
+                json!({})
+            );
+            assert!(!host.restart_requested());
+            assert!(host.0.restart_pending.load(Ordering::SeqCst));
+            assert_eq!(
+                host.call("core.restart", json!({})).await.unwrap(),
+                json!({})
+            );
+            timeout(Duration::from_secs(1), host.exit_requested())
+                .await
+                .unwrap();
+            // A duplicate after the trigger must not schedule a second task.
+            host.0.exit.send_replace(false);
+            assert_eq!(
+                host.call("core.restart", json!({})).await.unwrap(),
+                json!({})
+            );
+            assert!(
+                timeout(Duration::from_millis(300), host.exit_requested())
+                    .await
+                    .is_err()
+            );
+            host.stop().await;
+        })
+        .await
+        .unwrap();
+    }
+}

cli/tests/host/core.rs

@@ -1931,6 +1931,64 @@ async fn detached_processes_need_explicit_termination() {
     }
 }
 
+/// PROTOCOL 9: restart answers the detached caller, then uses normal stop and status 75.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn core_restart_cleanly_stops_detached_caller() {
+    tokio::time::timeout(std::time::Duration::from_secs(30), async {
+        let w = TestWorld::new();
+        std::fs::write(
+            w.0.join("config/boot.yaml"),
+            br#"[{"method":"probe.restart","params":{}}]"#,
+        )
+        .unwrap();
+        let output = tokio::process::Command::new(env!("CARGO_BIN_EXE_sausage"))
+            .env("SAUSAGE_LOOP", "1")
+            .arg("--headless")
+            .arg("--cache-dir")
+            .arg(concat!(env!("CARGO_TARGET_TMPDIR"), "/cache"))
+            .arg("run")
+            .arg(&w.0)
+            .stdin(std::process::Stdio::null())
+            .kill_on_drop(true)
+            .output()
+            .await
+            .unwrap();
+        assert_eq!(
+            output.status.code(),
+            Some(75),
+            "{}",
+            String::from_utf8_lossy(&output.stderr)
+        );
+        assert_eq!(
+            std::fs::read_to_string(w.0.join("state/restart-answer")).unwrap(),
+            "received"
+        );
+        assert_eq!(
+            std::fs::read_to_string(w.0.join("state/restart-stop")).unwrap(),
+            "persisted after EOF"
+        );
+        let db = rusqlite::Connection::open(w.0.join("state/core/journal.sqlite")).unwrap();
+        let count: i64 = db
+            .query_row(
+                "SELECT count(*) FROM journal WHERE line = 'restart stop hook finished'",
+                [],
+                |row| row.get(0),
+            )
+            .unwrap();
+        assert_eq!(count, 1);
+        let killed: i64 = db
+            .query_row(
+                "SELECT count(*) FROM journal WHERE line = 'kill'",
+                [],
+                |row| row.get(0),
+            )
+            .unwrap();
+        assert_eq!(killed, 0);
+    })
+    .await
+    .unwrap();
+}
+
 /// PROTOCOL 9: core.exit from boot ends the host executable with status 75.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn core_exit_sets_executable_status_75() {

cli/tests/plugins/probe/src/main.rs

@@ -447,6 +447,21 @@ fn dispatch(request: &Value, lines: &mut Lines) -> Option<Value> {
             burst(request);
             return None;
         }
+        "restart" => {
+            assert_eq!(
+                lock(&json!({"path":"/state/restart-stop"}), lines),
+                json!({"held":true})
+            );
+            assert_eq!(
+                outgoing_call(&json!({"method":"core.restart", "params":{}}), lines),
+                json!({})
+            );
+            std::fs::write("/state/restart-answer", "received").unwrap();
+            serve(request, lines);
+            std::fs::write("/state/restart-stop", "persisted after EOF").unwrap();
+            eprintln!("restart stop hook finished");
+            return None;
+        }
         "serve" => {
             serve(request, lines);
             return None;