feat(supervisor): stage gateway configuration snapshot delivery

architecture/gateway.md

@@ -543,6 +543,24 @@ successful create therefore yields an immediately usable provider; failures roll
 back the provider record. Service-account JSON and private keys remain gateway-side
 refresh bootstrap material; sandboxes receive minted access tokens instead.
 
+## Supervisor configuration routing
+
+Committed configuration mutations publish component and scope identifiers,
+never configuration payloads, to a bounded coalescing scheduler. The scheduler
+builds the latest full snapshot for each affected active sandbox. An async
+router owns session lookup, message sizing, sequence allocation, and enqueue.
+Its local implementation uses the process-local supervisor registry. A future
+HA implementation can resolve the gateway that owns a session and forward the
+same typed message without changing mutation handlers.
+
+Polling remains authoritative during the first rollout stage. Snapshot build,
+fanout, or enqueue failure cannot fail a mutation that already committed.
+Provider snapshots may contain credentials and must not be
+persisted or included in logs.
+
+See [sandbox configuration delivery](sandbox.md#supervisor-configuration-delivery)
+for bootstrap, revision, and supervisor application semantics.
+
 ## Supervisor Relay
 
 Sandbox workloads maintain an outbound supervisor session to the gateway. This

architecture/sandbox.md

@@ -471,6 +471,49 @@ the structured 403 and authors the narrowest rule. Mechanistically mapping L7
 would either over-broaden rules or require path-templating logic that rots
 quickly.
 
+## Supervisor Configuration Delivery
+
+The gateway and supervisor must implement the same internal supervisor protocol
+revision. Peers built before the handshake existed report revision zero and are
+accepted for one release with a warning and a counter, because sandboxes keep
+their supervisor binary until they are recreated. The gateway includes a configuration bootstrap when it accepts a
+`ConnectSupervisor` session and can send complete component replacements on the
+same stream after policy, settings, or provider state changes.
+Bootstrap construction does not gate the session while polling remains
+authoritative. These payloads describe the latest effective state rather than
+the mutation that produced it. The gateway assigns ordering sequences within
+each session and component, while each snapshot retains its own content
+revision.
+
+Bootstrap components are independent read projections, not one atomic database
+snapshot. The sandbox configuration carries the provider-environment revision
+it was built against. The gateway retries bootstrap construction when that
+revision does not match the provider snapshot. Later component updates and
+polling repair changes committed while the other projections were being built.
+
+Configuration delivery goes through a gateway-owned routing boundary rather
+than exposing local supervisor channels to mutation handlers. The current
+implementation routes only to a supervisor connected to the same gateway
+process. The asynchronous router contract can resolve a remote owner later
+without changing publishers. Provider payloads can contain
+credentials, so the gateway does not persist or render complete stream
+messages in logs.
+
+The supervisor currently parses and ignores stream-delivered configuration.
+Polling remains the only path that changes runtime state and repairs dropped or
+unavailable delivery. The gateway serializes construction per sandbox and
+component, and coalesces repeated mutations into the latest full snapshot. An
+enqueue result means only that the local stream queue accepted the message. A
+bounded scope fanout scheduler coalesces repeated workspace and global changes,
+and a semaphore sized from the database pool bounds how many snapshots build
+at once so a fleet-wide change cannot saturate the store or credential
+backends. Snapshot construction has a deadline that starts once a build holds a
+permit, and the gateway rejects encoded stream messages that approach the
+transport decoder limit. A later migration will apply these payloads directly
+and acknowledge their exact revisions before removing supervisor polling. At
+that point, the gateway will require a valid bootstrap before marking a session
+ready.
+
 ## Policy Revision Acknowledgement
 
 When the supervisor loads a sandbox-scoped policy from the gateway, it retains

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

@@ -73,3 +73,18 @@ pub use middleware::v1::*;
 pub use openshell::*;
 pub use sandbox::v1::*;
 pub use test::ObjectForTest;
+
+/// 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;

crates/openshell-server/Cargo.toml

@@ -122,6 +122,7 @@ test-support = []
 
 [dev-dependencies]
 base64 = { workspace = true }
+tokio = { workspace = true, features = ["test-util"] }
 hyper-rustls = { version = "0.27", default-features = false, features = ["native-tokio", "http1", "tls12", "logging", "aws-lc-rs"] }
 rcgen = { workspace = true }
 rsa = { version = "0.9", features = ["pem"] }

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

@@ -989,6 +989,18 @@ impl ComputeRuntime {
                 }
             })?;
 
+        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()
         {
@@ -1026,6 +1038,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())
@@ -1034,6 +1050,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())
@@ -1042,6 +1062,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())
@@ -4953,6 +4977,7 @@ pub async 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,
@@ -11255,6 +11280,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;