Use five-second upgrade grace for every plugin

Ignore legacy grace metadata in both activation paths. Keep superseded delivery,
admission ordering, discovery exclusion and the separate EOF stop budget.
Document the uniform policy and cover both windows and opt-in drain.

Card: SAU-357

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"]
-                    .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,
-                ));
+                upgrades.push(self.snapshot_upgrade(plugin, old.hash.clone(), &current.hash));
             }
         }
         self.publish_graph(plugins);

cli/src/upgrades.rs

@@ -4,6 +4,12 @@ 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,7 +19,6 @@ pub(crate) struct Upgrade {
     plugin: String,
     previous: String,
     current: String,
-    grace: Duration,
     processes: Vec<OldProcess>,
     services: Vec<(Service, Option<watch::Receiver<bool>>)>,
 }
@@ -24,7 +29,6 @@ impl Host {
         plugin: &str,
         previous: String,
         current: &str,
-        grace: f64,
     ) -> Upgrade {
         let mut state = self.0.state.lock().unwrap();
         let processes = state
@@ -69,7 +73,6 @@ impl Host {
             plugin: plugin.into(),
             previous,
             current: current.into(),
-            grace: Duration::from_secs_f64(grace),
             processes,
             services,
         }
@@ -92,7 +95,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,12 +105,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_stop(&process.pid); }
                     }
                 });
             }

cli/src/upgrades/tests.rs

@@ -0,0 +1,113 @@
+//! 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,
+        },
+    );
+    (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_stop 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;
+}