feat(supervisor): stage gateway configuration snapshot delivery

crates/openshell-cli/tests/ensure_providers_integration.rs

@@ -79,6 +79,13 @@ impl TestOpenShell {
 
 #[tonic::async_trait]
 impl OpenShell for TestOpenShell {
+    async fn peer_notify_config_update(
+        &self,
+        _request: tonic::Request<openshell_core::proto::PeerConfigUpdateHintRequest>,
+    ) -> Result<Response<openshell_core::proto::PeerConfigUpdateHintResponse>, Status> {
+        Err(Status::unimplemented("not used by this test server"))
+    }
+
     async fn peer_report_provider_readiness(
         &self,
         _request: tonic::Request<openshell_core::proto::ReportProviderReadinessRequest>,

crates/openshell-cli/tests/mtls_integration.rs

@@ -32,6 +32,13 @@ struct TestOpenShell;
 
 #[tonic::async_trait]
 impl OpenShell for TestOpenShell {
+    async fn peer_notify_config_update(
+        &self,
+        _request: tonic::Request<openshell_core::proto::PeerConfigUpdateHintRequest>,
+    ) -> Result<Response<openshell_core::proto::PeerConfigUpdateHintResponse>, Status> {
+        Err(Status::unimplemented("not used by this test server"))
+    }
+
     async fn peer_report_provider_readiness(
         &self,
         _request: tonic::Request<openshell_core::proto::ReportProviderReadinessRequest>,

crates/openshell-cli/tests/provider_commands_integration.rs

@@ -232,6 +232,13 @@ impl TestOpenShell {
 
 #[tonic::async_trait]
 impl OpenShell for TestOpenShell {
+    async fn peer_notify_config_update(
+        &self,
+        _request: tonic::Request<openshell_core::proto::PeerConfigUpdateHintRequest>,
+    ) -> Result<Response<openshell_core::proto::PeerConfigUpdateHintResponse>, Status> {
+        Err(Status::unimplemented("not used by this test server"))
+    }
+
     async fn peer_report_provider_readiness(
         &self,
         _request: tonic::Request<openshell_core::proto::ReportProviderReadinessRequest>,

crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs

@@ -92,6 +92,13 @@ struct TestOpenShell {
 
 #[tonic::async_trait]
 impl OpenShell for TestOpenShell {
+    async fn peer_notify_config_update(
+        &self,
+        _request: tonic::Request<openshell_core::proto::PeerConfigUpdateHintRequest>,
+    ) -> Result<Response<openshell_core::proto::PeerConfigUpdateHintResponse>, Status> {
+        Err(Status::unimplemented("not used by this test server"))
+    }
+
     async fn peer_report_provider_readiness(
         &self,
         _request: tonic::Request<openshell_core::proto::ReportProviderReadinessRequest>,

crates/openshell-cli/tests/sandbox_name_fallback_integration.rs

@@ -47,6 +47,13 @@ struct TestOpenShell {
 
 #[tonic::async_trait]
 impl OpenShell for TestOpenShell {
+    async fn peer_notify_config_update(
+        &self,
+        _request: tonic::Request<openshell_core::proto::PeerConfigUpdateHintRequest>,
+    ) -> Result<Response<openshell_core::proto::PeerConfigUpdateHintResponse>, Status> {
+        Err(Status::unimplemented("not used by this test server"))
+    }
+
     async fn peer_report_provider_readiness(
         &self,
         _request: tonic::Request<openshell_core::proto::ReportProviderReadinessRequest>,

crates/openshell-core/src/config.rs

@@ -46,6 +46,30 @@ pub const DEFAULT_DOCKER_NETWORK_NAME: &str = "openshell-docker";
 /// Default domain used for browser-facing sandbox service URLs.
 pub const DEFAULT_SERVICE_ROUTING_DOMAIN: &str = "openshell.localhost";
 
+/// Gateway delivery path for supervisor configuration.
+/// `Push` is a shadow stream until supervisors support applying snapshots.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ConfigDeliveryMode {
+    #[default]
+    Poll,
+    Push,
+}
+
+impl FromStr for ConfigDeliveryMode {
+    type Err = String;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        match value {
+            "poll" => Ok(Self::Poll),
+            "push" => Ok(Self::Push),
+            _ => Err(format!(
+                "invalid config delivery mode '{value}'; expected poll or push"
+            )),
+        }
+    }
+}
+
 /// Gateway posture when a sandbox rejects a candidate policy generation.
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
@@ -185,6 +209,9 @@ pub struct Config {
     /// Security posture for rejected sandbox policy generations.
     pub policy_validation_failure_mode: PolicyValidationFailureMode,
 
+    /// Optional shadow push of complete supervisor configuration snapshots.
+    pub config_delivery_mode: ConfigDeliveryMode,
+
     /// TLS configuration.  When `None`, the server listens on plaintext HTTP.
     pub tls: Option<TlsConfig>,
 
@@ -851,6 +878,7 @@ impl Config {
             metrics_bind_address: None,
             log_level: default_log_level(),
             policy_validation_failure_mode: PolicyValidationFailureMode::default(),
+            config_delivery_mode: ConfigDeliveryMode::default(),
             tls,
             oidc: None,
             auth: GatewayAuthConfig::default(),
@@ -1116,10 +1144,11 @@ const fn default_ssh_session_ttl_secs() -> u64 {
 #[cfg(test)]
 mod tests {
     use super::{
-        AppArmorProfile, Config, DEFAULT_SERVICE_ROUTING_DOMAIN, GatewayInterceptorBindingPolicy,
-        GatewayInterceptorConfig, GatewayInterceptorFailurePolicy, GatewayJwtConfig,
-        GatewayProviderProfileSourceConfig, ImagePullPolicy, PolicyValidationFailureMode,
-        UpstreamProxyConfig, default_sandbox_pids_limit, normalize_compute_driver_name,
+        AppArmorProfile, Config, ConfigDeliveryMode, DEFAULT_SERVICE_ROUTING_DOMAIN,
+        GatewayInterceptorBindingPolicy, GatewayInterceptorConfig, GatewayInterceptorFailurePolicy,
+        GatewayJwtConfig, GatewayProviderProfileSourceConfig, ImagePullPolicy,
+        PolicyValidationFailureMode, UpstreamProxyConfig, default_sandbox_pids_limit,
+        normalize_compute_driver_name,
     };
     use std::net::SocketAddr;
     use std::time::Duration;
@@ -1139,6 +1168,19 @@ mod tests {
         assert!("keep_old".parse::<PolicyValidationFailureMode>().is_err());
     }
 
+    #[test]
+    fn config_delivery_defaults_to_poll_and_rejects_unknown_modes() {
+        assert_eq!(
+            Config::new(None).config_delivery_mode,
+            ConfigDeliveryMode::Poll
+        );
+        assert_eq!(
+            "push".parse::<ConfigDeliveryMode>().unwrap(),
+            ConfigDeliveryMode::Push
+        );
+        assert!("enabled".parse::<ConfigDeliveryMode>().is_err());
+    }
+
     #[test]
     fn compute_driver_name_normalization_accepts_builtin_and_custom_names() {
         assert_eq!(normalize_compute_driver_name(" VM ").unwrap(), "vm");

crates/openshell-core/src/proto/mod.rs

@@ -99,6 +99,21 @@ pub fn all_workspaces_selector() -> WorkspaceSelector {
     }
 }
 
+/// Exact protocol revision required between a gateway and its supervisor.
+///
+/// The supervisor stream is an internal, version-locked deployment contract.
+/// Bump this when either peer can no longer honor the previous stream
+/// semantics.
+pub const SUPERVISOR_PROTOCOL_REVISION: u32 = 1;
+
+/// Revision implied by peers built before the handshake existed. Proto3 leaves
+/// the field unset, so such peers report zero.
+///
+/// Sandboxes keep their supervisor binary until they are recreated, so a
+/// gateway upgrade must keep serving them for one release. Remove this
+/// allowance once every supported release sends an explicit revision.
+pub const LEGACY_SUPERVISOR_PROTOCOL_REVISION: u32 = 0;
+
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;

crates/openshell-sdk/tests/client_mock.rs

@@ -229,6 +229,13 @@ fn workload_template_proto(name: &str, workspace: &str) -> proto::SandboxWorkloa
 
 #[tonic::async_trait]
 impl OpenShell for TestOpenShell {
+    async fn peer_notify_config_update(
+        &self,
+        _request: tonic::Request<proto::PeerConfigUpdateHintRequest>,
+    ) -> Result<Response<proto::PeerConfigUpdateHintResponse>, Status> {
+        Err(Status::unimplemented("not used by this test server"))
+    }
+
     async fn peer_report_provider_readiness(
         &self,
         _request: tonic::Request<proto::ReportProviderReadinessRequest>,

crates/openshell-server/src/auth/method_authz.rs

@@ -155,6 +155,7 @@ mod tests {
             "/openshell.v1.OpenShell/PeerReportProviderReadiness",
             "/openshell.v1.OpenShell/PeerReportEndpointStatus",
             "/openshell.v1.OpenShell/PeerGetSandboxProviderStatus",
+            "/openshell.v1.OpenShell/PeerNotifyConfigUpdate",
         ] {
             assert!(!is_user_callable(path));
             assert!(is_peer_callable(path));

crates/openshell-server/src/auth/peer.rs

@@ -715,6 +715,7 @@ mod tests {
             "/openshell.v1.OpenShell/PeerReportProviderReadiness",
             "/openshell.v1.OpenShell/PeerReportEndpointStatus",
             "/openshell.v1.OpenShell/PeerGetSandboxProviderStatus",
+            "/openshell.v1.OpenShell/PeerNotifyConfigUpdate",
         ] {
             assert!(matches!(
                 auth.authenticate(&headers, path).await.unwrap(),
@@ -727,7 +728,7 @@ mod tests {
                 .unwrap()
                 .is_none()
         );
-        assert_eq!(resolver.seen_tokens.lock().unwrap().len(), 4);
+        assert_eq!(resolver.seen_tokens.lock().unwrap().len(), 5);
     }
 
     fn identity() -> ResolvedGatewayPeerIdentity {

crates/openshell-server/src/cli.rs

@@ -6,7 +6,7 @@
 use clap::parser::ValueSource;
 use clap::{ArgAction, ArgMatches, Command, CommandFactory, FromArgMatches, Parser};
 use miette::{IntoDiagnostic, Result};
-use openshell_core::config::{DEFAULT_GATEWAY_NAME, DEFAULT_SERVER_PORT};
+use openshell_core::config::{ConfigDeliveryMode, DEFAULT_GATEWAY_NAME, DEFAULT_SERVER_PORT};
 use std::collections::BTreeMap;
 use std::ffi::OsString;
 use std::net::{IpAddr, SocketAddr};
@@ -239,6 +239,10 @@ struct RunArgs {
     #[arg(long, env = "OPENSHELL_GRPC_RATE_LIMIT_WINDOW_SECONDS")]
     grpc_rate_limit_window_seconds: Option<u64>,
 
+    /// Supervisor configuration delivery. Push sends shadow snapshots in this release.
+    #[arg(long, env = "OPENSHELL_CONFIG_DELIVERY_MODE")]
+    config_delivery_mode: Option<ConfigDeliveryMode>,
+
     /// Subject Alternative Names configured on the gateway server certificate.
     /// Wildcard DNS SANs also enable sandbox service URLs under that domain.
     #[arg(
@@ -550,6 +554,13 @@ fn prepare_server_config_with_drivers(
     {
         config.policy_validation_failure_mode = mode;
     }
+    config.config_delivery_mode = args
+        .config_delivery_mode
+        .or_else(|| {
+            file.as_ref()
+                .and_then(|f| f.openshell.gateway.config_delivery_mode)
+        })
+        .unwrap_or_default();
 
     if let Some(issuer) = args.oidc_issuer.clone() {
         config = config.with_oidc(openshell_core::OidcConfig {

crates/openshell-server/src/compute/mod.rs

@@ -1100,6 +1100,18 @@ impl ComputeRuntime {
         }
         drop(global_guard);
 
+        if let Err(status) = Box::pin(crate::grpc::policy::initialize_policy_history(
+            self.store.as_ref(),
+            &sandbox,
+            crate::grpc::policy::InitialPolicyHistoryStatus::Pending,
+        ))
+        .await
+        {
+            let _ = self.store.delete(Sandbox::object_type(), &sandbox_id).await;
+            self.sandbox_index.remove_sandbox(&sandbox_id);
+            return Err(status);
+        }
+
         if let Some(token) = sandbox_token
             && let Some(spec) = driver_sandbox.spec.as_mut()
         {
@@ -1178,6 +1190,10 @@ impl ComputeRuntime {
                 Ok(sandbox)
             }
             Err(status) if status.code() == Code::AlreadyExists => {
+                let _ = self
+                    .store
+                    .delete_by_scope(POLICY_OBJECT_TYPE, sandbox.object_id())
+                    .await;
                 let _ = self
                     .store
                     .delete(Sandbox::object_type(), sandbox.object_id())
@@ -1186,6 +1202,10 @@ impl ComputeRuntime {
                 Err(Status::already_exists("sandbox already exists"))
             }
             Err(status) if status.code() == Code::FailedPrecondition => {
+                let _ = self
+                    .store
+                    .delete_by_scope(POLICY_OBJECT_TYPE, sandbox.object_id())
+                    .await;
                 let _ = self
                     .store
                     .delete(Sandbox::object_type(), sandbox.object_id())
@@ -1194,6 +1214,10 @@ impl ComputeRuntime {
                 Err(Status::failed_precondition(status.message().to_string()))
             }
             Err(err) => {
+                let _ = self
+                    .store
+                    .delete_by_scope(POLICY_OBJECT_TYPE, sandbox.object_id())
+                    .await;
                 let _ = self
                     .store
                     .delete(Sandbox::object_type(), sandbox.object_id())
@@ -6941,6 +6965,7 @@ pub fn new_test_runtime_with_driver(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::policy_store::PolicyStoreExt;
     use futures::stream;
     use openshell_core::proto::compute::v1::{
         CreateSandboxResponse, DeleteSandboxResponse, GetCapabilitiesResponse, GetSandboxRequest,
@@ -15142,6 +15167,36 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn create_sandbox_persists_initial_policy_revision() {
+        let runtime = test_runtime(Arc::new(TestDriver::default())).await;
+        let mut sandbox = sandbox_record(
+            "sb-initial-policy",
+            "initial-policy",
+            SandboxPhase::Provisioning,
+        );
+        let policy = openshell_core::proto::SandboxPolicy::default();
+        sandbox.spec = Some(SandboxSpec {
+            policy: Some(policy.clone()),
+            ..Default::default()
+        });
+
+        runtime.create_sandbox(sandbox, None, false).await.unwrap();
+
+        let revision = runtime
+            .store
+            .get_latest_policy("sb-initial-policy")
+            .await
+            .unwrap()
+            .expect("initial policy revision");
+        assert_eq!(revision.version, 1);
+        assert_eq!(
+            revision.policy_hash,
+            crate::grpc::policy::deterministic_policy_hash(&policy)
+        );
+        assert_eq!(revision.status, "pending");
+    }
+
     #[tokio::test]
     async fn created_sandbox_is_immediately_visible_to_label_selectors() {
         let runtime = test_runtime(Arc::new(TestDriver::default())).await;