Merge '6649f4ca51a535632b3986c48e6f42ee0224d429'

PROTOCOL.md

@@ -19,7 +19,7 @@ Rust plugins use the SDK `sausage-plugin`. Its source is at
 | call | A JSON-RPC request. `notes.add` calls the method `add` of the plugin `notes`. |
 | interface | A set of methods. A plugin has one main interface. It can also declare endpoint interfaces. |
 | endpoint | An interface that a running process serves on a TCP port. An endpoint file tells others where it is. |
-| service | A call in `/config/boot.yaml`. The core starts it at boot and again after an upgrade of its plugin. |
+| service | A call in `/config/boot.yaml`. The core starts it at boot, restarts it after a crash, and starts it again after an upgrade of its plugin. |
 | import | An interface of another plugin that a plugin uses. |
 | provider, consumer | The plugin that has an interface, and a plugin that imports it. |
 | journal | The central log of the world. |
@@ -159,8 +159,11 @@ active (section 10.3):
    continue to work. It can stop accepting work, finish its in-flight calls,
    and exit early. Ignoring the notice changes nothing. The grace period is
    the upper bound before the core stops the process (section 3.6); its
-   default is 60 seconds. A plugin can declare its own value in its old
+   duration is 5 seconds for every plugin. Discovery metadata, including an
-   `rpc.discover` document (`x-sausage-grace`).
+   explicit `x-sausage-grace`, does not change it. After grace, stdin closes
+   and the separate 5-second stop budget applies (section 3.6): at most two
+   5-second windows. Discovery-only processes do not receive this notice or
+   enter upgrade grace; they follow the discovery lifetime (section 10.1).
 3. When the old process of a service has ended, the core starts the service
    again with the new version.
 4. When the last process of an older version has ended, the core starts a
@@ -353,10 +356,29 @@ At start, the core starts each service in `/config/boot.yaml`, in order:
 - A service is a detached call with the parent `core`. The core waits for
   its result before it starts the next one. The result shows that the
   service is ready.
-- If a service fails, the core writes the error to the journal and continues.
+- If a service fails before answering, the core writes the error to the
-- After an upgrade of the plugin of a service, the core starts the service
+  journal and continues booting. A failure to admit a process (for example,
-  again with the new version (section 3.7).
+  a missing plugin or invalid parameters) is logged, not retried in a loop.
-- The core does not restart other detached processes.
+- The core supervises the process lifetime, not its readiness response. A
+  nonzero exit, trap or failed host task restarts the same service with its
+  original method and parameters, including a crash before readiness. A
+  readiness response alone never triggers a restart; a clean exit does not
+  count as a crash or trigger an immediate restart.
+- More than 5 crashes within 10 seconds gives up on that service and writes
+  `service gave up` to the journal. Each service has an independent sliding
+  window; crashes exactly 10 seconds old expire. There is no retry delay or
+  periodic liveness check. Giving up is latched for this host lifetime:
+  neither elapsed time nor an upgrade re-arms it.
+- After an upgrade of the plugin of a service, the same supervisor starts
+  it again with the current version after its old process ends (section
+  3.7). Superseded exits are retirement, not crashes. Admission and cleanup
+  reserve a single owner/attempt, so an upgrade racing a crash cannot start
+  two owners. A matching boot-service start from `rpc.retired` is ignored
+  (returns `null`), even while restart is pending or after giving up; older
+  plugins' fallback starts do not bypass supervision.
+- An explicit `core.stop` or `core.kill` disables automatic attempts for
+  that service, including during an upgrade. Host shutdown starts no new
+  processes. The core does not restart other detached processes.
 - A singleton is a service with a fixed endpoint in `/run`. Only the core
   starts it, so two instances cannot start at the same time.
 
@@ -424,7 +446,8 @@ Params are by name. The document can have these extension fields:
   `methods`.
 - `x-sausage-imports`: for each provider, the part of its document that the
   plugin uses.
-- `x-sausage-grace`: the grace period in seconds.
+- `x-sausage-grace`: legacy grace declaration in seconds. The core ignores
+  this field; upgrade grace is always 5 seconds (section 3.7).
 
 The schema of params describes what the provider can read. The schema of a
 result describes what the provider writes. Descriptions are part of the

README.md

@@ -93,10 +93,16 @@ plugins/<name>/plugin.wasm   the current version of a plugin
 plugins/<name>/source/       its source
 state/<plugin>/              the state of a plugin, by convention
 state/core/journal.sqlite    the journal: stderr of all processes, and core events
-config/boot.yaml             the services that start with the world
+config/boot.yaml             ordered boot services, supervised for crashes (PROTOCOL §8)
 config/agent.yaml            the configuration of the agent
 ```
 
+Boot services restart on process crashes, not on readiness responses. More
+than 5 crashes within 10 seconds stops automatic attempts for that service
+and journals `service gave up`; an upgrade does not reset that decision.
+Explicitly stopping or killing a service also leaves it stopped for this
+host lifetime. See [Boot and services](PROTOCOL.md#8-boot-and-services).
+
 Rust builds use the SDK in `plugins/cargo/sdk/` directly. The host seeds this
 folder only when it is missing; restarts and host upgrades preserve existing
 files. SDK edits take effect on the next `cargo.build`, without a restart.

cli/src/activation.rs

@@ -279,15 +279,10 @@ impl Host {
                 .get(&candidate.plugin)
                 .filter(|old| !old.hash.is_empty() && old.hash != candidate.interface.hash)
             {
-                let grace = old.document["x-sausage-grace"]
-                    .as_f64()
-                    .filter(|value| value.is_finite() && *value >= 0.0)
-                    .unwrap_or(60.0);
                 upgrades.push(self.snapshot_upgrade(
                     &candidate.plugin,
                     old.hash.clone(),
                     &candidate.interface.hash,
-                    grace,
                 ));
             }
             self.0.journal.write(

cli/src/graph.rs

@@ -223,16 +223,7 @@ impl Host {
             if let Some(old) = old_plugins.get(plugin).filter(|old| {
                 !old.hash.is_empty() && old.hash != current.hash && current.error.is_none()
             }) {
-                let grace = old.document["x-sausage-grace"]
+                upgrades.push(self.snapshot_upgrade(plugin, old.hash.clone(), &current.hash));
-                    .as_f64()
-                    .filter(|value| value.is_finite() && *value >= 0.0)
-                    .unwrap_or(60.0);
-                upgrades.push(self.snapshot_upgrade(
-                    plugin,
-                    old.hash.clone(),
-                    &current.hash,
-                    grace,
-                ));
             }
         }
         self.publish_graph(plugins);

cli/src/process.rs

@@ -33,6 +33,8 @@ pub(crate) struct Process {
     pub(crate) id: Option<Value>,
     pub(crate) answer: Option<oneshot::Sender<RpcResult>>,
     pub(crate) ended: watch::Sender<bool>,
+    pub(crate) stopped: bool,
+    pub(crate) superseded: bool,
 }
 impl Host {
     /// Called only after the execution task ends. Wait for the store and file operations before releasing locks.
@@ -57,6 +59,7 @@ impl Host {
                 .map(|_| self.0.journal.query_later(json!({"pid":pid,"limit":20})));
             self.0.journal.forget(pid);
             Self::release_locks(&mut state, pid);
+            self.service_ended(&mut state, pid, status, p.stopped, p.superseded);
             let status = status.to_string();
             tokio::spawn(async move {
                 if let Some(answer) = p.answer {
@@ -84,13 +87,24 @@ impl Host {
     }
     /// Close stdin and start the same five-second deadline for each attached child.
     pub(crate) fn begin_stop(&self, pid: &str) -> Option<watch::Receiver<bool>> {
+        self.stop_process(pid, false)
+    }
+    /// Grace expiry is retirement, not an operator stopping the service.
+    pub(crate) fn begin_retire(&self, pid: &str) -> Option<watch::Receiver<bool>> {
+        self.stop_process(pid, true)
+    }
+    fn stop_process(&self, pid: &str, retirement: bool) -> Option<watch::Receiver<bool>> {
         let mut state = self.0.state.lock().unwrap();
         let p = state.processes.get_mut(pid)?;
+        if !retirement {
+            p.superseded = false;
+        }
         let ended = p.ended.subscribe();
         if p.info.state == "stopping" {
             return Some(ended);
         }
         p.info.state = "stopping".into();
+        p.stopped = true;
         p.input.take();
         self.0.journal.write(pid, &p.info.plugin, "core", "stop");
         for lock in state.locks.values_mut() {
@@ -110,17 +124,24 @@ impl Host {
         let pid = pid.to_string();
         tokio::spawn(async move {
             tokio::time::sleep(std::time::Duration::from_secs(5)).await;
-            h.kill(&pid);
+            h.kill_process(&pid, retirement);
         });
         Some(ended)
     }
     /// Abort a process task and perform its ended-state cleanup.
     pub(crate) fn kill(&self, pid: &str) {
+        self.kill_process(pid, false);
+    }
+    fn kill_process(&self, pid: &str, retirement: bool) {
         let mut state = self.0.state.lock().unwrap();
         for lock in state.locks.values_mut() {
             lock.queue.retain(|(queued, _)| queued != pid);
         }
         if let Some(p) = state.processes.get_mut(pid) {
+            if !retirement {
+                p.superseded = false;
+            }
+            p.stopped = true;
             p.info.state = "stopping".into();
             p.input.take();
             self.0.journal.write(pid, &p.info.plugin, "core", "kill");

cli/src/routing.rs

@@ -171,6 +171,33 @@ impl Host {
                 {
                     return Err(error(-32600, "caller stopped or ended"));
                 }
+                // Older plugins may still try to start their boot service from
+                // rpc.retired. The supervisor owns it even while a restart is
+                // pending, or after it has given up. Do not create a second owner.
+                if state
+                    .processes
+                    .get(&caller)
+                    .is_some_and(|p| p.info.method == "rpc.retired")
+                    && state
+                        .services
+                        .iter()
+                        .any(|s| s.method == format!("{plugin}.{method}") && s.params == params)
+                {
+                    return Ok(Value::Null);
+                }
+                if caller == "core"
+                    && let Some(service) = state
+                        .services
+                        .iter_mut()
+                        .find(|s| Some(&s.request) == id.as_ref())
+                {
+                    if !service.enabled || service.owner.is_some() {
+                        return Err(error(-32600, "service attempt is no longer current"));
+                    }
+                    service.owner = Some(pid.clone());
+                    service.pending = false;
+                    service.version = Some(info.version.clone());
+                }
                 host.0
                     .journal
                     .write(&pid, plugin, "core", &format!("start {method} {params}"));
@@ -187,6 +214,8 @@ impl Host {
                         id: id.clone(),
                         answer: if id.is_some() { Some(tx) } else { None },
                         ended: watch::channel(false).0,
+                        stopped: false,
+                        superseded: false,
                     },
                 );
             }

cli/src/services.rs

@@ -1,12 +1,36 @@
-//! Ordered boot services and their identities for upgrade restarts (PROTOCOL.md section 8).
+//! Ordered boot calls and the single restart owner for each service (PROTOCOL.md section 8).
-use crate::Host;
+use crate::{Host, host::State};
 use serde_json::{Value, json};
-#[derive(Clone)]
+use std::{collections::VecDeque, sync::atomic::Ordering, time::Duration};
+use tokio::time::Instant;
+
+// Agreed systemd-derived values: /docs/design/plugin-handover/design.md,
+// https://www.man7.org/linux/man-pages/man5/systemd-system.conf.5.html.
+const CRASH_LIMIT: usize = 5;
+const CRASH_WINDOW: Duration = Duration::from_secs(10);
+
 pub(crate) struct Service {
     pub(crate) method: String,
     pub(crate) params: Value,
     pub(crate) request: Value,
-    pub(crate) starting: std::sync::Arc<std::sync::atomic::AtomicBool>,
+    pub(crate) owner: Option<String>,
+    pub(crate) pending: bool,
+    pub(crate) version: Option<String>,
+    pub(crate) enabled: bool,
+    crashes: VecDeque<Instant>,
+}
+impl Service {
+    fn crashed(&mut self, now: Instant) -> bool {
+        while self
+            .crashes
+            .front()
+            .is_some_and(|t| now.duration_since(*t) >= CRASH_WINDOW)
+        {
+            self.crashes.pop_front();
+        }
+        self.crashes.push_back(now);
+        self.crashes.len() <= CRASH_LIMIT
+    }
 }
 impl Host {
     /// Starts discovery in the background, then waits for each boot service in order.
@@ -24,30 +48,119 @@ impl Host {
         }
         Ok(())
     }
-    /// Records a unique request identity before the service can answer or exit.
+    /// Register once, before admission or readiness. Restarts reuse this registration.
     pub(crate) async fn start_service(&self, method: String, params: Value) {
-        if self.0.shutdown.load(std::sync::atomic::Ordering::SeqCst) {
+        let request = json!(uuid::Uuid::now_v7().to_string());
+        {
+            let mut state = self.0.state.lock().unwrap();
+            if self.0.shutdown.load(Ordering::SeqCst) {
                 return;
             }
-        let request = json!(uuid::Uuid::now_v7().to_string());
+            state.services.push(Service {
-        let starting = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true));
+                method,
-        self.0.state.lock().unwrap().services.push(Service {
+                params,
-            method: method.clone(),
-            params: params.clone(),
                 request: request.clone(),
-            starting: starting.clone(),
+                owner: None,
+                enabled: true,
+                pending: true,
+                version: None,
+                crashes: VecDeque::new(),
             });
+        }
+        self.run_service(request).await;
+    }
+    /// Await readiness only; process cleanup, not this response, decides whether to restart.
+    async fn run_service(&self, request: Value) {
+        let (method, params) = {
+            let state = self.0.state.lock().unwrap();
+            if self.0.shutdown.load(Ordering::SeqCst) {
+                return;
+            }
+            let Some(service) = state
+                .services
+                .iter()
+                .find(|s| s.request == request && s.enabled)
+            else {
+                return;
+            };
+            (service.method.clone(), service.params.clone())
+        };
         self.0
             .journal
             .write("core", "core", "core", &format!("service start {method}"));
-        if let Err(error) = self
+        let result = self
-            .route("core".into(), Some(request), method, params, true)
+            .route("core".into(), Some(request.clone()), method, params, true)
-            .await
+            .await;
         {
+            let mut state = self.0.state.lock().unwrap();
+            if let Some(service) = state.services.iter_mut().find(|s| s.request == request) {
+                service.pending = false;
+            }
+        }
+        if let Err(error) = result {
+            // Admission failures have no process lifetime to supervise. A process crash
+            // before readiness is handled by finish just like one after readiness.
             self.0
                 .journal
                 .write("core", "core", "core", &format!("service failed {error}"));
         }
-        starting.store(false, std::sync::atomic::Ordering::SeqCst);
+    }
+    /// Called under the state lock after the old owner and its files/locks are gone.
+    pub(crate) fn service_ended(
+        &self,
+        state: &mut State,
+        pid: &str,
+        status: &str,
+        stopped: bool,
+        superseded: bool,
+    ) {
+        let Some(service) = state
+            .services
+            .iter_mut()
+            .find(|s| s.owner.as_deref() == Some(pid))
+        else {
+            return;
+        };
+        service.owner = None;
+        if self.0.shutdown.load(Ordering::SeqCst) || !service.enabled {
+            return;
+        }
+        if !superseded {
+            if stopped {
+                service.enabled = false;
+                return;
+            }
+            if status == "exit 0" {
+                return;
+            }
+            if !service.crashed(Instant::now()) {
+                service.enabled = false;
+                self.0.journal.write(
+                    "core",
+                    "core",
+                    "core",
+                    &format!(
+                        "service gave up {}: more than {CRASH_LIMIT} crashes within {} seconds",
+                        service.method,
+                        CRASH_WINDOW.as_secs(),
+                    ),
+                );
+                return;
+            }
+        }
+        self.queue_service(service);
+    }
+    /// Reserve the next attempt before spawning it; upgrades and crashes share this owner.
+    pub(crate) fn queue_service(&self, service: &mut Service) {
+        service.request = json!(uuid::Uuid::now_v7().to_string());
+        service.pending = true;
+        let request = service.request.clone();
+        let host = self.clone();
+        tokio::spawn(async move {
+            host.run_service(request).await;
+        });
     }
 }
+
+#[cfg(test)]
+mod tests;

cli/src/services/tests.rs

@@ -0,0 +1,356 @@
+//! Deterministic service supervision tests: native tasks stand in for guest lifetimes.
+use super::*;
+use crate::{NativePlugin, runtime::Runtime};
+use std::sync::Arc;
+use tokio::sync::{mpsc, oneshot, watch};
+
+struct Rig {
+    host: Host,
+    starts: mpsc::UnboundedReceiver<(String, oneshot::Sender<()>)>,
+    _world: tempfile::TempDir,
+}
+impl Rig {
+    async fn new() -> Self {
+        let world = tempfile::tempdir().unwrap();
+        let runtime = Runtime::new(world.path()).unwrap();
+        let host = Host::test_world(world.path(), runtime).await.unwrap();
+        let (tx, starts) = mpsc::unbounded_channel();
+        let plugin: NativePlugin = Arc::new(move |ctx, method, _| {
+            let tx = tx.clone();
+            Box::pin(async move {
+                if method != "serve" {
+                    return Ok(Value::Null);
+                }
+                let (end, ended) = oneshot::channel();
+                tx.send((ctx.pid, end)).unwrap();
+                let _ = ended.await;
+                Ok(Value::Null)
+            })
+        });
+        host.register_native("demo", plugin.clone());
+        host.register_native("other", plugin);
+        Self {
+            host,
+            starts,
+            _world: world,
+        }
+    }
+    async fn start(&mut self, method: &str) -> (String, oneshot::Sender<()>) {
+        let host = self.host.clone();
+        let method = method.to_string();
+        tokio::spawn(async move {
+            host.start_service(method, json!({"kept":true})).await;
+        });
+        self.next().await
+    }
+    async fn next(&mut self) -> (String, oneshot::Sender<()>) {
+        // Reuse the host integration call bound (activation.rs::call).
+        tokio::time::timeout(Duration::from_secs(60), self.starts.recv())
+            .await
+            .unwrap()
+            .unwrap()
+    }
+    fn ended(&self, pid: &str) -> watch::Receiver<bool> {
+        self.host.0.state.lock().unwrap().processes[pid]
+            .ended
+            .subscribe()
+    }
+    async fn crash(&self, pid: &str) {
+        let mut ended = self.ended(pid);
+        // Abort execution without asking the supervisor for an intentional stop.
+        self.host.0.state.lock().unwrap().processes[pid]
+            .abort
+            .as_ref()
+            .unwrap()
+            .abort();
+        ended.wait_for(|done| *done).await.unwrap();
+    }
+    async fn clean(&self, pid: &str, end: oneshot::Sender<()>) {
+        let mut ended = self.ended(pid);
+        end.send(()).unwrap();
+        ended.wait_for(|done| *done).await.unwrap();
+    }
+    fn assert_single(&self, pid: &str, registrations: usize) {
+        let state = self.host.0.state.lock().unwrap();
+        assert_eq!(state.services.len(), registrations);
+        let p = &state.processes[pid];
+        assert_eq!(p.info.params, json!({"kept":true}));
+        assert!(p.info.detached);
+        assert_eq!(p.caller, "core");
+        assert_eq!(
+            state
+                .services
+                .iter()
+                .filter(|s| s.owner.as_deref() == Some(pid))
+                .count(),
+            1
+        );
+    }
+    fn no_start(&mut self) {
+        assert!(self.starts.try_recv().is_err());
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn crash_before_and_after_readiness_observes_process_not_response() {
+    let mut rig = Rig::new().await;
+    let (pid, _end) = rig.start("demo.serve").await;
+    rig.assert_single(&pid, 1);
+    rig.crash(&pid).await; // No ready RPC response was sent.
+    let (pid, end) = rig.next().await;
+    rig.assert_single(&pid, 1);
+    rig.host.answer(&pid, Ok(json!({"ready":true})));
+    tokio::task::yield_now().await;
+    rig.no_start(); // Ready does not end the service lifetime.
+    rig.crash(&pid).await;
+    drop(end);
+    let (pid, end) = rig.next().await;
+    rig.assert_single(&pid, 1);
+    rig.clean(&pid, end).await;
+    rig.no_start(); // A clean, intentional return is not a crash.
+    rig.host.stop().await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn fifth_restarts_sixth_gives_up_and_journals_once() {
+    let mut rig = Rig::new().await;
+    let (mut pid, mut end) = rig.start("demo.serve").await;
+    for crash in 1..=CRASH_LIMIT {
+        rig.crash(&pid).await;
+        drop(end);
+        (pid, end) = rig.next().await;
+        rig.assert_single(&pid, 1);
+        assert_eq!(
+            rig.host.0.state.lock().unwrap().services[0].crashes.len(),
+            crash
+        );
+    }
+    rig.crash(&pid).await;
+    drop(end);
+    rig.no_start();
+    let logs = rig
+        .host
+        .call("core.log", json!({"grep":"service gave up"}))
+        .await
+        .unwrap();
+    assert_eq!(logs.as_array().unwrap().len(), 1);
+    assert!(
+        logs[0]["line"]
+            .as_str()
+            .unwrap()
+            .contains("more than 5 crashes within 10 seconds")
+    );
+    assert!(!rig.host.0.state.lock().unwrap().services[0].enabled);
+    // A retired hook cannot bypass give-up while its supervisor has no owner.
+    rig.host.register_native(
+        "fallback",
+        Arc::new(move |ctx, _, _| {
+            Box::pin(async move {
+                ctx.call(
+                    "core.call",
+                    json!({"method":"demo.serve","params":{"kept":true},"detach":true}),
+                )
+                .await
+            })
+        }),
+    );
+    assert_eq!(
+        rig.host
+            .call("fallback.rpc.retired", json!({}))
+            .await
+            .unwrap(),
+        Value::Null
+    );
+    rig.no_start();
+    // Neither time passing nor an upgrade re-arms a given-up service.
+    tokio::time::advance(CRASH_WINDOW).await;
+    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
+    rig.host.begin_upgrade(upgrade);
+    tokio::task::yield_now().await;
+    rig.no_start();
+    rig.host.stop().await;
+}
+
+#[test]
+fn sliding_window_boundary_and_progress() {
+    let now = Instant::now();
+    let service = || Service {
+        method: "demo.serve".into(),
+        params: json!({}),
+        request: json!("request"),
+        owner: None,
+        enabled: true,
+        pending: false,
+        version: None,
+        crashes: VecDeque::new(),
+    };
+    let mut before = service();
+    let mut exact = service();
+    let mut after = service();
+    for _ in 0..CRASH_LIMIT {
+        assert!(before.crashed(now));
+        assert!(exact.crashed(now));
+        assert!(after.crashed(now));
+    }
+    assert!(!before.crashed(now + CRASH_WINDOW - Duration::from_nanos(1)));
+    assert!(exact.crashed(now + CRASH_WINDOW));
+    assert_eq!(exact.crashes.len(), 1);
+    assert!(after.crashed(now + CRASH_WINDOW + Duration::from_nanos(1)));
+    // Repeated progress beyond the window never accumulates old failures.
+    for step in 2..=CRASH_LIMIT as u32 + 1 {
+        assert!(exact.crashed(now + CRASH_WINDOW * step));
+        assert_eq!(exact.crashes.len(), 1);
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn services_have_independent_crash_budgets() {
+    let mut rig = Rig::new().await;
+    let (mut a, mut a_end) = rig.start("demo.serve").await;
+    let (b, b_end) = rig.start("other.serve").await;
+    for _ in 0..CRASH_LIMIT {
+        rig.crash(&a).await;
+        drop(a_end);
+        (a, a_end) = rig.next().await;
+    }
+    rig.crash(&a).await;
+    drop(a_end);
+    rig.crash(&b).await;
+    drop(b_end);
+    let (b, b_end) = rig.next().await;
+    rig.assert_single(&b, 2);
+    rig.clean(&b, b_end).await;
+    rig.no_start();
+    rig.host.stop().await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn upgrade_crash_race_has_one_restart_and_retired_fallback_cannot_duplicate_it() {
+    let mut rig = Rig::new().await;
+    let (pid, end) = rig.start("demo.serve").await;
+    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
+    rig.crash(&pid).await; // Exits between snapshot and begin_upgrade.
+    drop(end);
+    rig.host.begin_upgrade(upgrade);
+    let (pid, end) = rig.next().await;
+    rig.assert_single(&pid, 1);
+    assert!(
+        rig.host.0.state.lock().unwrap().services[0]
+            .crashes
+            .is_empty()
+    );
+    // A native retired hook exercises the same routing path as a guest fallback.
+    rig.host.register_native(
+        "fallback",
+        Arc::new(move |ctx, _, _| {
+            Box::pin(async move {
+                ctx.call(
+                    "core.call",
+                    json!({"method":"demo.serve","params":{"kept":true},"detach":true}),
+                )
+                .await
+            })
+        }),
+    );
+    assert_eq!(
+        rig.host
+            .call("fallback.rpc.retired", json!({}))
+            .await
+            .unwrap(),
+        Value::Null
+    );
+    rig.no_start();
+    rig.clean(&pid, end).await;
+    rig.host.stop().await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn crash_before_upgrade_snapshot_also_has_one_restart() {
+    let mut rig = Rig::new().await;
+    let (pid, end) = rig.start("demo.serve").await;
+    rig.crash(&pid).await;
+    drop(end);
+    // Restart may be pending admission or already admitted: snapshot handles both.
+    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "native");
+    rig.host.begin_upgrade(upgrade);
+    let (pid, end) = rig.next().await;
+    rig.assert_single(&pid, 1);
+    rig.no_start();
+    rig.clean(&pid, end).await;
+    rig.host.stop().await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn pending_admission_is_single_owned_across_upgrade() {
+    let mut rig = Rig::new().await;
+    let (pid, end) = rig.start("demo.serve").await;
+    let host = rig.host.clone();
+    let admission = host.0.admission.write().await;
+    rig.crash(&pid).await;
+    drop(end);
+    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "native");
+    rig.host.begin_upgrade(upgrade);
+    assert!(rig.host.0.state.lock().unwrap().services[0].pending);
+    drop(admission);
+    let (pid, end) = rig.next().await;
+    rig.assert_single(&pid, 1);
+    rig.no_start();
+    rig.clean(&pid, end).await;
+    rig.host.stop().await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn shutdown_rejects_restart_already_waiting_for_admission() {
+    let mut rig = Rig::new().await;
+    let (pid, end) = rig.start("demo.serve").await;
+    let host = rig.host.clone();
+    let admission = host.0.admission.write().await;
+    rig.crash(&pid).await;
+    drop(end);
+    rig.host.stop().await;
+    drop(admission);
+    tokio::task::yield_now().await;
+    rig.no_start();
+    assert!(rig.host.processes().is_empty());
+}
+
+#[tokio::test(start_paused = true)]
+async fn clean_service_stays_dormant_until_upgrade_and_explicit_stop_stays_disabled() {
+    let mut rig = Rig::new().await;
+    let (pid, end) = rig.start("demo.serve").await;
+    rig.clean(&pid, end).await;
+    rig.no_start();
+    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
+    rig.host.begin_upgrade(upgrade);
+    let (pid, end) = rig.next().await;
+    rig.assert_single(&pid, 1);
+    // An explicit stop stays disabled even if the task subsequently exits cleanly.
+    let mut ended = rig.host.begin_stop(&pid).unwrap();
+    end.send(()).unwrap();
+    ended.wait_for(|done| *done).await.unwrap();
+    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "newer");
+    rig.host.begin_upgrade(upgrade);
+    tokio::task::yield_now().await;
+    rig.no_start();
+    rig.host.stop().await;
+}
+
+#[tokio::test(start_paused = true)]
+async fn intentional_stop_and_shutdown_never_restart_even_during_upgrade() {
+    let mut rig = Rig::new().await;
+    let (pid, end) = rig.start("demo.serve").await;
+    let upgrade = rig.host.snapshot_upgrade("demo", "old".into(), "new");
+    // An explicit kill after supersession overrides the automatic upgrade restart.
+    rig.host
+        .call("core.kill", json!({"pid":pid}))
+        .await
+        .unwrap();
+    drop(end);
+    rig.host.begin_upgrade(upgrade);
+    tokio::task::yield_now().await;
+    rig.no_start();
+    let (_pid, end) = rig.start("other.serve").await;
+    rig.host.stop().await;
+    drop(end);
+    rig.no_start();
+}

cli/src/upgrades.rs

@@ -1,9 +1,15 @@
 //! Grace periods, service restarts, and retirement of replaced versions (PROTOCOL.md section 3.7).
-use crate::{Host, services::Service};
+use crate::Host;
 use serde_json::json;
 use std::{collections::BTreeMap, sync::atomic::Ordering, time::Duration};
 use tokio::sync::watch;
 
+/// Uniform host policy (PROTOCOL 3.7), independent of discovery metadata.
+const UPGRADE_GRACE: Duration = Duration::from_secs(5);
+
+#[cfg(test)]
+mod tests;
+
 pub(crate) struct OldProcess {
     pid: String,
     version: String,
@@ -13,9 +19,7 @@ pub(crate) struct Upgrade {
     plugin: String,
     previous: String,
     current: String,
-    grace: Duration,
     processes: Vec<OldProcess>,
-    services: Vec<(Service, Option<watch::Receiver<bool>>)>,
 }
 impl Host {
     /// Snapshots old admissions while activation holds the admission write guard.
@@ -24,54 +28,48 @@ impl Host {
         plugin: &str,
         previous: String,
         current: &str,
-        grace: f64,
     ) -> Upgrade {
         let mut state = self.0.state.lock().unwrap();
         let processes = state
             .processes
-            .iter()
+            .iter_mut()
             .filter(|(_, process)| {
                 process.info.plugin == plugin
                     && process.info.version != current
                     && process.info.method != "rpc.discover"
             })
-            .map(|(pid, process)| OldProcess {
+            .map(|(pid, process)| {
+                // The service supervisor owns the restart, including an exit on
+                // the superseded notice before the grace timer is installed.
+                process.superseded = !process.stopped;
+                OldProcess {
                     pid: pid.clone(),
                     version: process.info.version.clone(),
                     ended: process.ended.subscribe(),
+                }
             })
             .collect();
-        let mut services = Vec::new();
+        // A cleanly returned service is dormant until an upgrade. A pending
-        let mut retained = Vec::new();
+        // admission will choose the current version itself. Disabled services
-        for service in std::mem::take(&mut state.services) {
+        // (explicit stop or crash-limit give-up) must never be re-armed here.
-            if service
+        for service in &mut state.services {
+            if service.enabled
+                && !service.pending
+                && service.owner.is_none()
+                && service.version.as_deref() != Some(current)
+                && service
                     .method
                     .split_once('.')
                     .is_some_and(|(name, _)| name == plugin)
             {
-                let process = state
+                self.queue_service(service);
-                    .processes
-                    .values()
-                    .find(|process| process.id.as_ref() == Some(&service.request));
-                if process.is_some_and(|process| process.info.version == current)
-                    || (process.is_none() && service.starting.load(Ordering::SeqCst))
-                {
-                    retained.push(service);
-                } else {
-                    services.push((service, process.map(|process| process.ended.subscribe())));
-                }
-            } else {
-                retained.push(service);
             }
         }
-        state.services = retained;
         Upgrade {
             plugin: plugin.into(),
             previous,
             current: current.into(),
-            grace: Duration::from_secs_f64(grace),
             processes,
-            services,
         }
     }
     /// Lets old calls finish, stops them after grace, and starts each lifecycle hook once.
@@ -92,7 +90,7 @@ impl Host {
                 "core",
                 &upgrade.plugin,
                 "core",
-                &format!("upgrade grace {} seconds", upgrade.grace.as_secs_f64()),
+                &format!("upgrade grace {} seconds", UPGRADE_GRACE.as_secs()),
             );
             let mut versions: BTreeMap<String, Vec<watch::Receiver<bool>>> = BTreeMap::new();
             versions.entry(upgrade.previous).or_default();
@@ -102,23 +100,11 @@ impl Host {
                     .or_default()
                     .push(process.ended.clone());
                 let host = host.clone();
-                let grace = upgrade.grace;
                 tokio::spawn(async move {
                     let mut ended = process.ended;
                     tokio::select! {
                         _ = ended.wait_for(|ended|*ended) => (),
-                        _ = tokio::time::sleep(grace) => { host.begin_stop(&process.pid); }
+                        _ = tokio::time::sleep(UPGRADE_GRACE) => { host.begin_retire(&process.pid); }
-                    }
-                });
-            }
-            for (service, ended) in upgrade.services {
-                let host = host.clone();
-                tokio::spawn(async move {
-                    if let Some(mut ended) = ended {
-                        let _ = ended.wait_for(|ended| *ended).await;
-                    }
-                    if !host.0.shutdown.load(Ordering::SeqCst) {
-                        host.start_service(service.method, service.params).await;
                     }
                 });
             }

cli/src/upgrades/tests.rs

@@ -0,0 +1,115 @@
+//! Controlled-clock coverage of both upgrade windows and discovery exclusion.
+use super::*;
+use crate::{
+    process::{Process, ProcessInfo},
+    runtime::Runtime,
+    world::World,
+};
+use tokio::{
+    sync::mpsc,
+    task::JoinHandle,
+    time::{Instant, advance},
+};
+
+/// A process with active work that ignores both superseded and EOF.
+fn process(
+    host: &Host,
+    pid: &str,
+    method: &str,
+) -> (mpsc::UnboundedReceiver<serde_json::Value>, JoinHandle<()>) {
+    let (input, incoming) = mpsc::unbounded_channel();
+    let task = tokio::spawn(std::future::pending());
+    host.0.state.lock().unwrap().processes.insert(
+        pid.into(),
+        Process {
+            info: ProcessInfo {
+                plugin: "notes".into(),
+                version: "old".into(),
+                method: method.into(),
+                params: json!({}),
+                parent: None,
+                detached: true,
+                start_time: 0,
+                state: "running".into(),
+            },
+            memory: sausage_vfs::Memory::default(),
+            files: sausage_vfs::Process::default(),
+            store_ended: watch::channel(true).0,
+            input: Some(input),
+            abort: Some(task.abort_handle()),
+            caller: "console".into(),
+            id: None,
+            answer: None,
+            ended: watch::channel(false).0,
+            stopped: false,
+            superseded: false,
+        },
+    );
+    (incoming, task)
+}
+
+/// PROTOCOL 3.7/3.6: five seconds running, then five seconds stopping; discovery is untouched.
+#[tokio::test]
+async fn two_five_second_windows_and_discovery_exclusion() {
+    let dir = tempfile::tempdir().unwrap();
+    let root = dir.path().join("world");
+    World::create(&root).unwrap();
+    let host = Host::test_world(&root, Runtime::new(&dir.path().join("cache")).unwrap())
+        .await
+        .unwrap();
+    let (mut input, work) = process(&host, "old-work", "work");
+    let (mut discovery_input, discovery) = process(&host, "old-discovery", "rpc.discover");
+    let upgrade = host.snapshot_upgrade("notes", "old".into(), "new");
+    assert_eq!(upgrade.processes.len(), 1);
+    tokio::time::pause();
+    let start = Instant::now();
+    host.begin_upgrade(upgrade);
+    assert_eq!(
+        input.recv().await.unwrap(),
+        json!({"jsonrpc":"2.0","method":"rpc.superseded","params":{"current":"new"}})
+    );
+    // Run the upgrade dispatcher, then its per-process task, so the deadline is armed.
+    tokio::task::yield_now().await;
+    tokio::task::yield_now().await;
+    let window = Duration::from_secs(5);
+    assert_eq!(UPGRADE_GRACE, window);
+    // tokio::time::sleep documents millisecond granularity; the paused driver
+    // may round a deadline up by one tick. Check each window independently.
+    let tick = Duration::from_millis(1);
+    let before_deadline = window - tick;
+    advance(before_deadline).await;
+    tokio::task::yield_now().await;
+    assert_eq!(host.processes()["old-work"].state, "running");
+    assert!(!work.is_finished());
+    advance(tick).await;
+    assert!(
+        input.recv().await.is_none(),
+        "EOF at grace, even with active work"
+    );
+    let stopped = Instant::now();
+    assert!((window..=window + tick).contains(&(stopped - start)));
+    assert_eq!(host.processes()["old-work"].state, "stopping");
+    // begin_retire schedules the independent PROTOCOL 3.6 deadline.
+    tokio::task::yield_now().await;
+    advance(before_deadline).await;
+    tokio::task::yield_now().await;
+    assert!(
+        !work.is_finished(),
+        "on_stop still has its full five seconds"
+    );
+    advance(tick).await;
+    assert!(work.await.unwrap_err().is_cancelled());
+    assert!((window..=window + tick).contains(&(Instant::now() - stopped)));
+    assert_eq!(host.processes()["old-discovery"].state, "running");
+    assert!(matches!(
+        discovery_input.try_recv(),
+        Err(mpsc::error::TryRecvError::Empty)
+    ));
+    assert!(!discovery.is_finished());
+    discovery.abort();
+    let _ = discovery.await;
+    // Synthetic processes have no execution cleanup task; discard only this test table.
+    host.0.shutdown.store(true, Ordering::SeqCst);
+    host.0.state.lock().unwrap().processes.clear();
+    host.stop().await;
+}

cli/tests/host/activation.rs

@@ -160,9 +160,9 @@ impl Drop for TestWorld {
         let _ = std::fs::remove_dir_all(&self.0);
     }
 }
-/// Builds a method-only document without an upgrade delay.
+/// Builds a method-only document; all plugins use the host's five-second grace.
 fn document(name: &str) -> Value {
-    json!({"methods":[{"name":name,"params":[],"result":{"schema":{"type":"string"}}}],"x-sausage-grace":0})
+    json!({"methods":[{"name":name,"params":[],"result":{"schema":{"type":"string"}}}]})
 }
 /// Adds an import of a provider's method list.
 fn consumer(provider: &str, expected: Value) -> Value {
@@ -484,7 +484,7 @@ async fn review_dropped_activation_completes() {
     host.stop().await;
 }
 /// PROTOCOL 10.1: discovery fails after its fixed ten-second deadline and receives EOF.
-/// PROTOCOL 3.7: old processes retain their declared grace while new calls use new code.
+/// PROTOCOL 3.7: old processes get five seconds regardless of declarations; new calls use new code.
 /// PROTOCOL 3.7: retirement waits for all live admissions, including reverted versions.
 /// PROTOCOL 3.7: rejected discovery candidates never receive retirement hooks.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -604,7 +604,7 @@ async fn source_changes_do_not_change_candidate() {
     );
     host.stop().await;
 }
-/// PROTOCOL 3.7: plugins without a declared grace period retain old calls before the default deadline.
+/// PROTOCOL 3.7: missing metadata also retains old calls until the uniform deadline.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn default_grace_retains_old_process() {
     let world = TestWorld::new();
@@ -623,6 +623,104 @@ async fn default_grace_retains_old_process() {
     assert_eq!(host.processes()[&pid].state, "running");
     host.stop().await;
 }
+/// PROTOCOL 3.7: metadata cannot shorten or extend the actual grace for active requests.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn uniform_grace_ignores_discovery_declarations() {
+    let world = TestWorld::new();
+    let declarations = [
+        ("missing", None),
+        ("default", Some(json!(60))),
+        ("long", Some(json!(3600))),
+        ("zero", Some(json!(0))),
+        ("negative", Some(json!(-1))),
+        ("invalid", Some(json!("long"))),
+        ("huge", Some(json!(1e300))),
+    ];
+    for (name, grace) in &declarations {
+        let mut old = document("work");
+        if let Some(grace) = grace {
+            old["x-sausage-grace"] = grace.clone();
+        }
+        world.plugin(name, old, document("work"));
+    }
+    let host = world.start().await;
+    let mut ready = host.subscribe();
+    let mut calls = Vec::new();
+    for (name, _) in &declarations {
+        let host = host.clone();
+        let method = format!("{name}.work");
+        calls.push(tokio::spawn(async move {
+            let result = host.call(&method, json!({})).await;
+            (result, std::time::Instant::now())
+        }));
+    }
+    // Guest notifications are events, not process/file polling.
+    for _ in &declarations {
+        let event = tokio::time::timeout(Duration::from_secs(30), ready.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(event["params"]["name"], "ready");
+    }
+    let before = std::time::Instant::now();
+    let plugins: Vec<_> = declarations
+        .iter()
+        .map(|(name, _)| json!({"plugin":name,"wasm":format!("/state/{name}/next.wasm")}))
+        .collect();
+    call(&host, "core.activate", json!({"plugins":plugins})).await;
+    // Use the existing discovery+stop test budget, not a new execution limit.
+    tokio::time::timeout(Duration::from_secs(15), async {
+        for task in calls {
+            let (result, ended) = task.await.unwrap();
+            assert_eq!(result.unwrap_err()["code"], -32002);
+            assert!(
+                ended.duration_since(before) >= Duration::from_secs(5),
+                "every declaration gets the full grace"
+            );
+        }
+    })
+    .await
+    .unwrap();
+    host.stop().await;
+}
+
+/// PROTOCOL 10.4/3.7: a watcher refresh uses the same policy as explicit activation.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn manual_copy_uses_uniform_grace() {
+    let world = TestWorld::new();
+    let mut old = document("work");
+    old["x-sausage-grace"] = json!(3600);
+    world.plugin("notes", old, document("work"));
+    let host = world.start().await;
+    // Wait for initial discovery without changing the installed version.
+    call(
+        &host,
+        "core.activate",
+        json!({"plugin":"notes","wasm":"/plugins/notes/plugin.wasm"}),
+    )
+    .await;
+    let mut ready = host.subscribe();
+    let work_host = host.clone();
+    let work = tokio::spawn(async move { work_host.call("notes.work", json!({})).await });
+    let event = tokio::time::timeout(Duration::from_secs(30), ready.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(event["params"]["name"], "ready");
+    let before = std::time::Instant::now();
+    std::fs::copy(fixture(2), world.0.join("plugins/notes/plugin.wasm")).unwrap();
+    assert_eq!(
+        tokio::time::timeout(Duration::from_secs(15), work)
+            .await
+            .unwrap()
+            .unwrap()
+            .unwrap_err()["code"],
+        -32002
+    );
+    assert!(before.elapsed() >= Duration::from_secs(5));
+    host.stop().await;
+}
+
 /// PROTOCOL 10.3: candidate imports must fit installed providers.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn checks_candidate_imports() {
@@ -818,11 +916,11 @@ async fn graph_missing_import() {
     assert!(dot.as_str().unwrap().contains("absent"));
     host.stop().await;
 }
-/// Checks the declared upgrade grace and dispatch to the new version.
+/// Checks that an explicit long declaration cannot extend grace; new calls use the new version.
 async fn old_process_grace_and_new_calls() {
     let world = TestWorld::new();
     let mut old = document("add");
-    old["x-sausage-grace"] = json!(0.5);
+    old["x-sausage-grace"] = json!(3600);
     world.plugin("notes", old, document("add"));
     let host = world.start().await;
     call(&host, "notes.hold", json!({})).await;
@@ -868,7 +966,7 @@ async fn superseded_notifies_each_old_process() {
         json!({"jsonrpc":"2.0","method":"rpc.superseded","params":{"current":fixture_hash(2)}})
             .to_string();
     for pid in pids {
-        // marker's existing 30-second bound is shorter than the old 60-second grace.
+        // The notice arrives while the old process is still in its five-second window.
         marker(
             world.0.join(format!("state/notes/superseded-{pid}")),
             &expected,
@@ -993,7 +1091,7 @@ async fn upgrades_enter_the_journal() {
                 .as_array()
                 .unwrap()
                 .iter()
-                .any(|e| e["line"] == "upgrade grace 60 seconds")
+                .any(|e| e["line"] == "upgrade grace 5 seconds")
             {
                 break entries;
             }
@@ -1117,6 +1215,36 @@ async fn old_consumer_can_call_removed_provider_method() {
     host.stop().await;
 }
 
+/// PROTOCOL 8: supervise actual WASI exits on either side of the readiness response.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn crashed_boot_service_restarts_before_and_after_ready() {
+    for when in ["before", "after"] {
+        let world = TestWorld::new();
+        world.plugin("notes", document("serve"), document("serve"));
+        std::fs::write(
+            world.0.join("config/boot.yaml"),
+            json!([{"method":"notes.serve","params":{"crash_once":when}}]).to_string(),
+        )
+        .unwrap();
+        let host = world.start().await;
+        marker(world.0.join("state/notes/service"), "1").await;
+        let entries = call(
+            &host,
+            "core.log",
+            json!({"grep":"service start notes.serve"}),
+        )
+        .await;
+        assert_eq!(entries.as_array().unwrap().len(), 2);
+        let processes = host.processes();
+        let services: Vec<_> = processes.values().filter(|p| p.method == "serve").collect();
+        assert_eq!(services.len(), 1);
+        assert_eq!(services[0].params, json!({"crash_once":when}));
+        let exits = call(&host, "core.log", json!({"plugin":"notes","grep":"exit 1"})).await;
+        assert_eq!(exits.as_array().unwrap().len(), 1);
+        host.stop().await;
+    }
+}
+
 /// PROTOCOL 10.2: activation and revert check interfaces but leave persistent state unchanged.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn state_is_outside_interface_checks() {

cli/tests/host/sdk.rs

@@ -178,6 +178,56 @@ async fn stop_runs_hook() {
     assert_eq!(std::fs::read(world.0.join("stopped")).unwrap(), b"yes");
     host.stop().await;
 }
+/// PROTOCOL 3.7: an idle SDK listener opts into drain and exits before EOF/grace.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn superseded_idle_listener_drains_before_grace() {
+    let world = TestWorld::new();
+    let host = world.start().await;
+    let (notice, mut notices) = tokio::sync::mpsc::unbounded_channel();
+    host.register_native(
+        "other",
+        std::sync::Arc::new(move |_, method, _| {
+            let notice = notice.clone();
+            Box::pin(async move {
+                assert_eq!(method, "superseded");
+                notice.send(std::time::Instant::now()).unwrap();
+                Ok(json!({}))
+            })
+        }),
+    );
+    let pid = call(&host, "probe.drain_service", json!({})).await;
+    // Replace with a different real component; force only skips unrelated test consumers.
+    std::fs::copy(
+        fixtures().join("notes.wasm"),
+        world.0.join("state/next.wasm"),
+    )
+    .unwrap();
+    call(
+        &host,
+        "core.activate",
+        json!({"plugin":"probe","wasm":"/state/next.wasm","force":true}),
+    )
+    .await;
+    // Activation preparation/discovery is outside upgrade grace. Time from the
+    // hook's outgoing call, not from the start of candidate compilation.
+    let before = tokio::time::timeout(Duration::from_secs(60), notices.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    // Waiting on its lifetime lock is an exit event, not process-liveness polling.
+    call(&host, "core.lock", json!({"path":"/drain-owner"})).await;
+    assert!(before.elapsed() < Duration::from_secs(5));
+    assert!(!host.processes().contains_key(pid.as_str().unwrap()));
+    assert!(
+        !world
+            .0
+            .join(format!("stopped-{}", pid.as_str().unwrap()))
+            .exists(),
+        "drain must not run on_stop (discovery processes may run their own hooks)"
+    );
+    host.stop().await;
+}
+
 /// SDK framing carries payloads above 64 KiB in both directions.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn large_messages() {

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

@@ -84,8 +84,31 @@ fn main() {
             .unwrap();
             json!({"retired":version})
         }
+        "work" => {
+            // Keep an actual request unanswered through grace; notices do not drain it.
+            send(
+                json!({"jsonrpc":"2.0","method":"core.notify","params":{"name":"ready","params":{}}}),
+            );
+            for _ in lines {}
+            return;
+        }
         "serve" | "hold" => {
+            // One genuine process failure, followed by a healthy service, without
+            // timers or liveness polling in either the fixture or supervisor.
+            let crash = request["params"]["crash_once"].as_str();
+            let first = crash.is_some()
+                && std::fs::OpenOptions::new()
+                    .write(true)
+                    .create_new(true)
+                    .open(format!("/state/{plugin}/crashed-once"))
+                    .is_ok();
+            if first && crash == Some("before") {
+                std::process::exit(1);
+            }
             send(json!({"jsonrpc":"2.0","id":request["id"],"result":{"version":version}}));
+            if first && crash == Some("after") {
+                std::process::exit(1);
+            }
             if request["method"] == "serve" {
                 std::fs::write(format!("/state/{plugin}/service"), version).unwrap();
             }

sdk/src/lib.rs

@@ -356,7 +356,9 @@
 //!
 //! The interface check covers calls, not state. Old and new processes share files in `/state`.
 //! `core.revert` restores code, not state. Preserve compatibility or migrate the state.
-//! `#[rpc(grace_seconds = N)]` declares the grace period for old processes. The default is 60 seconds.
+//! Core gives every old process 5 seconds of upgrade grace, then closes stdin and allows
+//! a separate 5-second stop budget before killing it. Discovery-only processes are excluded.
+//! `#[rpc(grace_seconds = N)]` still emits legacy `x-sausage-grace` metadata, but core ignores it.
 //! New calls use the new version immediately. Old processes can still use old imports during their grace period.
 //! [`macro@on_superseded`] handles the notice at grace start, with normal calls still available.
 //! It is opt-in: without a hook the process continues as before until core stops it.

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

@@ -128,6 +128,11 @@ impl Probe {
         sausage_plugin::listen("/state/control", std::rc::Rc::new(Control)).await?;
         Ok(sausage_plugin::pid())
     }
+    /// Keep an idle listener and ownership lock until opt-in drain ends this process.
+    async fn drain_service(&self) -> Result<String> {
+        lock("/drain-owner").await?.keep();
+        self.service().await
+    }
     /// Add two numbers.
     ///
     /// The result is their sum.
@@ -238,6 +243,7 @@ impl Probe {
     #[sausage_plugin::on_stop]
     async fn stop(&self) {
         std::fs::write("/stopped", b"yes").unwrap();
+        std::fs::write(format!("/stopped-{}", sausage_plugin::pid()), b"yes").unwrap();
         Ctx::default().notify("stopped", json!({})).unwrap();
         assert!(call("other.no", json!({})).await.is_err());
     }