Notify replaced plugin processes at upgrade grace start

Send rpc.superseded with the newly active hash before activation returns.
Add an opt-in SDK hook and drain helper that closes SDK listeners and idle
connections without cancelling active handlers or background work.
Document the additive notice and preserve existing grace, stop and restart rules.

Cover immediate notices, ignored notices, early service restarts, SDK draining
and duplicate-hook diagnostics.

Card: SAU-352

cli/src/upgrades.rs

@@ -12,6 +12,7 @@ pub(crate) struct OldProcess {
 pub(crate) struct Upgrade {
     plugin: String,
     previous: String,
+    current: String,
     grace: Duration,
     processes: Vec<OldProcess>,
     services: Vec<(Service, Option<watch::Receiver<bool>>)>,
@@ -67,6 +68,7 @@ impl Host {
         Upgrade {
             plugin: plugin.into(),
             previous,
+            current: current.into(),
             grace: Duration::from_secs_f64(grace),
             processes,
             services,
@@ -74,6 +76,13 @@ impl Host {
     }
     /// Lets old calls finish, stops them after grace, and starts each lifecycle hook once.
     pub(crate) fn begin_upgrade(&self, upgrade: Upgrade) {
+        // Queue the notice before returning from activation, without waiting on guest code.
+        for process in &upgrade.processes {
+            self.send(
+                &process.pid,
+                json!({"jsonrpc":"2.0","method":"rpc.superseded","params":{"current":upgrade.current}}),
+            );
+        }
         let host = self.clone();
         tokio::spawn(async move {
             if host.0.shutdown.load(Ordering::SeqCst) {

cli/tests/host/activation.rs

@@ -839,6 +839,76 @@ async fn old_process_grace_and_new_calls() {
         "2"
     );
     marker(world.0.join("state/notes/ended-1"), "ended").await;
+    let notice: Value = serde_json::from_slice(
+        &std::fs::read(world.0.join(format!("state/notes/superseded-{pid}"))).unwrap(),
+    )
+    .unwrap();
+    assert_eq!(notice["params"]["current"], fixture_hash(2));
+    host.stop().await;
+}
+/// PROTOCOL 3.7: every old process gets the new hash while still running, before grace ends.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn superseded_notifies_each_old_process() {
+    let world = TestWorld::new();
+    let mut old = document("hold");
+    old["x-sausage-grace"] = json!(60);
+    world.plugin("notes", old, document("hold"));
+    let host = world.start().await;
+    call(&host, "notes.hold", json!({})).await;
+    call(&host, "notes.hold", json!({})).await;
+    let pids: Vec<_> = host
+        .processes()
+        .into_iter()
+        .filter(|(_, process)| process.method == "hold")
+        .map(|(pid, _)| pid)
+        .collect();
+    assert_eq!(pids.len(), 2);
+    activate(&host, "notes").await.unwrap();
+    let expected =
+        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.
+        marker(
+            world.0.join(format!("state/notes/superseded-{pid}")),
+            &expected,
+        )
+        .await;
+        assert_eq!(host.processes()[&pid].state, "running");
+    }
+    assert_eq!(
+        call(&host, "notes.version", json!({})).await["version"],
+        "2"
+    );
+    host.stop().await;
+}
+/// PROTOCOL 3.7: a service that exits on the notice restarts before its grace deadline.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn superseded_service_exits_early() {
+    let world = TestWorld::new();
+    let mut old = document("serve");
+    old["x-sausage-grace"] = json!(60);
+    world.plugin("notes", old, document("serve"));
+    std::fs::write(
+        world.0.join("config/boot.yaml"),
+        json!([{"method":"notes.serve","params":{"exit_on_superseded":true}}]).to_string(),
+    )
+    .unwrap();
+    let host = world.start().await;
+    marker(world.0.join("state/notes/service"), "1").await;
+    let pid = host
+        .processes()
+        .into_iter()
+        .find(|(_, p)| p.method == "serve")
+        .unwrap()
+        .0;
+    activate(&host, "notes").await.unwrap();
+    marker(world.0.join("state/notes/service"), "2").await;
+    assert!(!host.processes().contains_key(&pid));
+    assert!(
+        !world.0.join("state/notes/ended-1").exists(),
+        "old service must exit on the notice, not EOF"
+    );
     host.stop().await;
 }
 /// PROTOCOL 3.7: retirement receives the previous hash only after old processes end.

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

@@ -89,7 +89,24 @@ fn main() {
             if request["method"] == "serve" {
                 std::fs::write(format!("/state/{plugin}/service"), version).unwrap();
             }
-            for _ in lines {}
+            for line in lines {
+                let Ok(message) = serde_json::from_str::<Value>(&line.unwrap()) else {
+                    continue;
+                };
+                if message["method"] == "rpc.superseded" {
+                    std::fs::write(
+                        format!(
+                            "/state/{plugin}/superseded-{}",
+                            std::env::var("SAUSAGE_PID").unwrap()
+                        ),
+                        message.to_string(),
+                    )
+                    .unwrap();
+                    if request["params"]["exit_on_superseded"] == true {
+                        return;
+                    }
+                }
+            }
             std::fs::write(format!("/state/{plugin}/ended-{version}"), "ended").unwrap();
             return;
         }