Add supervised restart support to Gateway backend

crates/gateway/app/src/admin/open/progress-tests.rs

@@ -9,7 +9,7 @@ use gateway_api_types::Progress;
 use gateway_progress::ProgressHub;
 
 use super::{PROGRESS_HEARTBEAT, progress_sse_response};
-use crate::shutdown::ShutdownSignal;
+use crate::shutdown::{ShutdownReason, ShutdownSignal};
 
 const FRAME_TIMEOUT: Duration = Duration::from_secs(5);
 
@@ -54,7 +54,7 @@ async fn a_fresh_subscriber_first_receives_the_current_snapshot() {
 
     // The subscriber connects after the work began: the stream must open
     // with the current state rather than wait for the next change.
-    let response = progress_sse_response(&hub, ShutdownSignal::default());
+    let response = progress_sse_response(&hub, Arc::new(ShutdownSignal::default()));
     let mut frames = response.into_body().into_data_stream();
     let snapshots = read_snapshots(&mut frames, 1).await;
     assert_eq!(snapshots[0], snapshot(true, "Downloading qwen 45%"));
@@ -64,7 +64,7 @@ async fn a_fresh_subscriber_first_receives_the_current_snapshot() {
 #[tokio::test]
 async fn an_idle_hub_opens_the_stream_with_the_idle_snapshot() {
     let hub = Arc::new(ProgressHub::new());
-    let response = progress_sse_response(&hub, ShutdownSignal::default());
+    let response = progress_sse_response(&hub, Arc::new(ShutdownSignal::default()));
     let mut frames = response.into_body().into_data_stream();
     let snapshots = read_snapshots(&mut frames, 1).await;
     assert_eq!(
@@ -77,7 +77,7 @@ async fn an_idle_hub_opens_the_stream_with_the_idle_snapshot() {
 #[tokio::test]
 async fn the_stream_sends_one_line_per_change_in_order() {
     let hub = Arc::new(ProgressHub::new());
-    let response = progress_sse_response(&hub, ShutdownSignal::default());
+    let response = progress_sse_response(&hub, Arc::new(ShutdownSignal::default()));
     let mut frames = response.into_body().into_data_stream();
     // Consume the opening idle snapshot.
     let opening = read_snapshots(&mut frames, 1).await;
@@ -103,7 +103,7 @@ async fn the_stream_sends_one_line_per_change_in_order() {
 #[tokio::test(start_paused = true)]
 async fn an_idle_hub_emits_heartbeat_comments_on_cadence() {
     let hub = Arc::new(ProgressHub::new());
-    let response = progress_sse_response(&hub, ShutdownSignal::default());
+    let response = progress_sse_response(&hub, Arc::new(ShutdownSignal::default()));
     let mut frames = response.into_body().into_data_stream();
     let opening = read_snapshots(&mut frames, 1).await;
     assert_eq!(opening[0], Progress::default());
@@ -126,7 +126,7 @@ async fn an_idle_hub_emits_heartbeat_comments_on_cadence() {
 #[tokio::test(start_paused = true)]
 async fn an_activity_drop_reports_idle_then_the_stream_goes_quiet() {
     let hub = Arc::new(ProgressHub::new());
-    let response = progress_sse_response(&hub, ShutdownSignal::default());
+    let response = progress_sse_response(&hub, Arc::new(ShutdownSignal::default()));
     let mut frames = response.into_body().into_data_stream();
     let opening = read_snapshots(&mut frames, 1).await;
     assert_eq!(opening[0], Progress::default());
@@ -153,7 +153,7 @@ async fn an_activity_drop_reports_idle_then_the_stream_goes_quiet() {
 #[tokio::test]
 async fn the_stream_ends_when_the_shutdown_signal_fires() {
     let hub = Arc::new(ProgressHub::new());
-    let shutdown = ShutdownSignal::default();
+    let shutdown = Arc::new(ShutdownSignal::default());
     let response = progress_sse_response(&hub, shutdown.clone());
     let mut frames = response.into_body().into_data_stream();
 
@@ -163,7 +163,7 @@ async fn the_stream_ends_when_the_shutdown_signal_fires() {
     let snapshots = read_snapshots(&mut frames, 1).await;
     assert_eq!(snapshots[0], snapshot(true, "Downloading qwen"));
 
-    shutdown.fire();
+    shutdown.fire(ShutdownReason::Stop);
     let end = tokio::time::timeout(FRAME_TIMEOUT, frames.next())
         .await
         .expect("the stream reacts to the signal within the frame timeout");

crates/gateway/app/src/admin/open/progress.rs

@@ -1,6 +1,8 @@
 //! The `GET /admin/progress` route: the process activity hub as an SSE
 //! stream of [`Progress`] snapshots with heartbeats.
 
+use std::sync::Arc;
+
 use axum::Router;
 use axum::body::Body;
 use axum::extract::State;
@@ -55,7 +57,7 @@ async fn admin_progress(
 /// Builds the progress SSE response over `hub`: the current snapshot first,
 /// then one line per change, heartbeats in the gaps, until `shutdown`
 /// fires.
-fn progress_sse_response(hub: &ProgressHub, shutdown: shutdown::ShutdownSignal) -> Response {
+fn progress_sse_response(hub: &ProgressHub, shutdown: Arc<shutdown::ShutdownSignal>) -> Response {
     // The receiver starts holding the current snapshot as unseen, so the
     // first `changed()` resolves at once with the opening line and nothing
     // between subscribe and first poll is lost.

crates/gateway/app/src/admin/open/status-tests.rs

@@ -293,3 +293,18 @@ async fn the_cancel_routes_fire_the_token_and_drop_pending_entries() {
     queue.shutdown();
     worker.await.expect("the worker exits on shutdown");
 }
+
+#[tokio::test]
+async fn the_status_response_reports_whether_a_supervisor_runs_the_gateway() {
+    let unsupervised = get_status(state()).await;
+    assert_eq!(unsupervised["supervised"], false);
+
+    let supervised_state = {
+        let mut state = state();
+        state.supervised = true;
+        state
+    };
+
+    let supervised = get_status(supervised_state).await;
+    assert_eq!(supervised["supervised"], true);
+}

crates/gateway/app/src/admin/open/status.rs

@@ -41,6 +41,9 @@ struct StatusReply {
     /// The process-lifetime identifier the config UI uses to detect a
     /// restart.
     config_generation: String,
+    /// Whether a supervisor runs this gateway, so `POST /admin/restart` is
+    /// available and the config UI offers Restart to Apply.
+    supervised: bool,
     /// The active profile's `models` allowlist, when it declared one.
     model_allowlist: Option<Vec<String>>,
     /// Running local children; zero in a headless build.
@@ -154,6 +157,7 @@ async fn admin_status(
         models,
         loading_models: live.loading.iter().cloned().collect(),
         config_generation: state.config_generation.to_string(),
+        supervised: state.supervised,
         model_allowlist: live.model_allowlist.clone(),
         local_children,
         vram_gb,

crates/gateway/app/src/admin/walled.rs

@@ -30,6 +30,7 @@ pub(crate) mod hf;
 mod model_info;
 #[cfg(feature = "local")]
 mod orphans;
+pub(crate) mod restart;
 pub(crate) mod reveal;
 pub(crate) mod shutdown;
 pub(crate) mod system;
@@ -46,6 +47,7 @@ use crate::registry::RouteInfo;
 pub(crate) fn routes() -> Router<AppState> {
     let router = Router::new()
         .merge(shutdown::routes())
+        .merge(restart::routes())
         .merge(system::routes())
         .merge(config::routes())
         .merge(config_pending::routes())
@@ -79,6 +81,7 @@ pub(crate) fn registry() -> Vec<RouteInfo> {
     )]
     let mut routes = [
         shutdown::ROUTES,
+        restart::ROUTES,
         system::ROUTES,
         config::ROUTES,
         config_pending::ROUTES,

crates/gateway/app/src/admin/walled/config_apply-tests-rollback.rs

@@ -0,0 +1,80 @@
+//! Apply keeps rollback copies: the first Apply since the last start copies
+//! each real file it replaces to `<file>.prev`, and later Applies keep that
+//! copy; an env-only change that breaks a `${VAR}` is refused.
+
+use gateway_config::write_shadow;
+
+use super::{CONFIG, GAMMA_MODEL, fixture, post, serve_fixture, stage_gamma};
+use crate::commands::apply::capture_apply;
+use crate::config_shadow::rollback_path;
+
+#[tokio::test]
+async fn apply_keeps_the_booted_config_as_the_rollback_copy() {
+    let (_temp, config, paths) = fixture();
+    let config_path = paths.config_path.clone();
+    stage_gamma(&config_path);
+    let (addr, _state) = serve_fixture(config, paths).await;
+
+    let response = post(addr, "admin/config-apply").await;
+
+    assert_eq!(response.status(), reqwest::StatusCode::OK);
+    assert_eq!(
+        std::fs::read_to_string(rollback_path(&config_path))
+            .expect("should be able to read the rollback copy"),
+        CONFIG,
+        "the rollback config should have booted"
+    );
+}
+
+#[tokio::test]
+async fn a_second_apply_keeps_the_first_rollback_copy() {
+    let (_temp, config, paths) = fixture();
+
+    let config_path = paths.config_path.clone();
+    stage_gamma(&config_path);
+
+    let (addr, _state) = serve_fixture(config, paths).await;
+    post(addr, "admin/config-apply").await;
+
+    let edited = format!("{CONFIG}{GAMMA_MODEL}")
+        .replace("description = \"gamma\"", "description = \"gamma, edited\"");
+
+    write_shadow(&config_path, &edited).expect("the second config change should succeed");
+
+    let response = post(addr, "admin/config-apply").await;
+
+    assert_eq!(response.status(), reqwest::StatusCode::OK);
+    assert_eq!(
+        std::fs::read_to_string(rollback_path(&config_path))
+            .expect("the rollback copy should be successfully read"),
+        CONFIG,
+        "a second call to Apply should not change the contents of the rollback file"
+    );
+}
+
+#[test]
+fn an_env_change_that_breaks_a_reference_is_refused() {
+    let temp = tempfile::TempDir::new().expect("temp dir");
+    let config_path = temp.path().join("gateway.toml");
+
+    std::fs::write(
+        &config_path,
+        "config-version = 0\n\n[server]\nbind = \"127.0.0.1:0\"\n\
+         api_key = \"${PF_TEST_ROLLBACK_KEY}\"\n",
+    )
+    .expect("write config");
+
+    let env_path = config_path.with_extension("env");
+
+    std::fs::write(&env_path, "PF_TEST_ROLLBACK_KEY=booted\n")
+        .expect("writing env file should succeed");
+
+    write_shadow(&env_path, "OTHER=value\n").expect("writing to shadow should succeed");
+
+    let result = capture_apply(&config_path);
+
+    assert!(
+        result.is_err(),
+        "an env-only change must resolve every reference the config makes"
+    );
+}

crates/gateway/app/src/admin/walled/config_apply-tests.rs

@@ -10,6 +10,9 @@ use crate::test_support::{AdminPaths, app_state, serve_state};
 #[path = "config_apply-tests-in-flight.rs"]
 mod in_flight;
 
+#[path = "config_apply-tests-rollback.rs"]
+mod rollback;
+
 const CONFIG: &str = r#"
 config-version = 0
 

crates/gateway/app/src/admin/walled/config_apply.rs

@@ -34,8 +34,8 @@ use tokio_util::sync::CancellationToken;
 use crate::AppState;
 use crate::auth::LoopbackCaller;
 use crate::commands::Command;
-use crate::commands::apply::{ApplyPlan, capture_apply, promote_captures};
-use crate::config_shadow::{config_root, relative_name, shadow_census};
+use crate::commands::apply::{ApplyPlan, capture_apply, promote_captures, write_rollback_files};
+use crate::config_shadow::{ShadowCensus, config_root, relative_name};
 use crate::error::error_chain;
 use crate::error::{GatewayError, blocking};
 use crate::registry::RouteInfo;
@@ -110,14 +110,22 @@ async fn admin_config_apply(
                 files,
                 restart_required,
             } => {
-                let applied = blocking(move || promote_captures(&files)).await??;
+                let applied = blocking(move || {
+                    write_rollback_files(&files)?;
+                    promote_captures(&files)
+                })
+                .await??;
                 return Ok(Json(ApplyReply {
                     applied,
                     reloaded: false,
                     restart_required,
                 }));
             }
-            ApplyPlan::Reload(snapshot) => snapshot,
+            ApplyPlan::Reload(snapshot) => {
+                let captures = snapshot.captures().to_vec();
+                blocking(move || write_rollback_files(&captures)).await??;
+                snapshot
+            }
         };
         let applied = snapshot.applied_names();
         let restart_required = snapshot.restart_required;
@@ -174,10 +182,10 @@ async fn admin_config_revert(
 /// Deletes every shadow the census finds, returning the deleted shadow
 /// files relative to the config root, sorted.
 fn delete_all_shadows(config_path: &Path) -> Result<Vec<String>, GatewayError> {
-    let census = shadow_census(config_path)?;
+    let census = ShadowCensus::new(config_path)?;
     let root = config_root(config_path);
-    let mut reverted: Vec<String> = Vec::with_capacity(census.files.len());
-    for file in &census.files {
+    let mut reverted: Vec<String> = Vec::with_capacity(census.files().len());
+    for file in census.files() {
         let shadow: PathBuf = shadow_path(file);
         std::fs::remove_file(&shadow)
             .map_err(|source| GatewayError::ConfigWriteIo(Box::new(source)))?;

crates/gateway/app/src/admin/walled/config_pending-tests.rs

@@ -126,3 +126,41 @@ async fn pending_view_reports_the_state_files_name_even_when_stale() {
         "the raw persisted name is reported even when the config no longer defines it"
     );
 }
+
+#[test]
+fn an_env_shadow_requires_a_restart() {
+    let temp = tempfile::TempDir::new().expect("creating a temp dir should succeed");
+    let config = temp.path().join("gateway.toml");
+    std::fs::write(
+        &config,
+        "config-version = 0\n[server]\nbind = \"127.0.0.1:0\"\napi_key = \"k\"\n",
+    )
+    .expect("writing config file should succeed");
+    write_shadow(&config.with_extension("env"), "PF_TEST_KEY=value\n")
+        .expect("staging env shadow should succeed");
+
+    let reply = dirty_reply(&config).expect("generating a dirty reply should succeed");
+
+    assert!(reply.restart_required);
+}
+
+#[test]
+fn a_live_reloadable_change_needs_no_restart() {
+    let temp = tempfile::TempDir::new().expect("creating a temp dir should succeed");
+    let config = temp.path().join("gateway.toml");
+    let base = "config-version = 0\n[server]\nbind = \"127.0.0.1:0\"\napi_key = \"k\"\n";
+    std::fs::write(&config, base).expect("write config");
+    write_shadow(
+        &config,
+        &format!(
+            "{base}\n[[endpoint]]\nid = \"fake\"\nprotocol = \"openai\"\n\
+             base_url = \"http://127.0.0.1:9\"\napi_key = \"\"\n"
+        ),
+    )
+    .expect("writing a shadow file should not fail");
+
+    let reply = dirty_reply(&config).expect("generating a dirty reply should succeed");
+
+    assert!(reply.dirty);
+    assert!(!reply.restart_required);
+}