@@ -379,6 +379,11 @@ deadline. Each retry re-reads the owner record, so a supervisor reconnect or
 heartbeat can surface a new owner; if no fresh reachable owner appears before
 the deadline, the client operation fails rather than electing an owner itself.
 
+Nothing redistributes established sessions, so after a rolling restart the last
+surviving replica holds most sessions and a new replica serves none until
+sandboxes reconnect. That skew decays only as sandboxes churn. Client traffic
+stays correct throughout because a non-owner relays to the owner.
+
 File upload and download use tar-over-SSH through the same relay path. A gateway
 pod termination drops the active SSH proxy byte stream, so the CLI retries the
 whole sync operation with a fresh SSH session instead of attempting mid-stream
@@ -393,10 +398,12 @@ also trust the chart CA, present the chart-generated client certificate for
 mTLS, and verify the stable gateway Service DNS name even when connecting to a
 Deployment pod IP.
 
-`WatchSandbox` uses the local update bus for same-replica writes. One shared
-poller per gateway observes resource-version changes made by other replicas and
-feeds that bus for all local watchers, avoiding a database poll per client
-stream.
+`WatchSandbox` uses the local update bus for same-replica writes. On
+multi-replica backends one shared poller per gateway observes resource-version
+changes made by other replicas and feeds that bus for all local watchers,
+avoiding a database poll per client stream. SQLite deployments do not run the
+poller because they are single-replica and the local bus already sees every
+write.
 
 Mutations whose invariants span sandbox, provider-profile, policy, or provider
 records take a process-local mutex and a shared PostgreSQL advisory lock. The
 
@@ -3378,9 +3378,9 @@ impl ComputeRuntime {
         else {
             return Ok(None);
         };
-        let age_ms = openshell_core::time::now_ms() - owner.updated_at_ms;
-        let ttl_ms = i64::try_from(OWNER_TTL.as_millis()).unwrap_or(i64::MAX);
-        Ok((age_ms < ttl_ms).then_some(owner.supervisor_instance_id))
+        Ok(owner
+            .is_fresh(OWNER_TTL)
+            .then_some(owner.supervisor_instance_id))
     }
 
     async fn set_supervisor_session_state(
@@ -4028,8 +4028,9 @@ impl ComputeRuntime {
             }
 
             let sandbox = decode_sandbox_record(&current_record)?;
-            let age_ms =
-                openshell_core::time::now_ms().saturating_sub(current_record.created_at_ms);
+            let age_ms = openshell_core::time::now_ms()
+                .saturating_sub(current_record.created_at_ms)
+                .max(0);
             if age_ms < grace_ms {
                 return Ok(());
             }
 
@@ -492,6 +492,27 @@ fn derive_peer_endpoint(config: &Config) -> Option<String> {
     ))
 }
 
+/// Reject a plaintext peer endpoint on a gateway that serves TLS.
+///
+/// Peer relay traffic carries whole supervisor sessions between replicas. The
+/// chart renders a plaintext peer endpoint only when the gateway itself serves
+fn validate_peer_endpoint_scheme(config: &Config, peer_endpoint: &str) -> Result<()> {
+    if peer_endpoint.starts_with("https://") {
+        return Ok(());
+    }
+    if config.tls.is_some() {
+        return Err(Error::config(format!(
+            "gateway peer endpoint {peer_endpoint} is plaintext but this gateway serves TLS; \
+             set an https:// OPENSHELL_PEER_ENDPOINT so peer relay traffic is not downgraded"
+        )));
+    }
+    warn!(
+        peer_endpoint,
+        "gateway peer relay traffic is plaintext because this gateway does not serve TLS"
+    );
+    Ok(())
+}
+
 /// Run the `OpenShell` server.
 ///
 /// This starts a multiplexed gRPC/HTTP server on the configured bind address.
@@ -753,6 +774,20 @@ pub(crate) async fn run_server(
         );
     }
 
+    let peer_routing_expected = state.peer_endpoint.is_some() && !state.store.is_single_replica();
+    if let Some(peer_endpoint) = state.peer_endpoint.as_deref()
+        && peer_routing_expected
+    {
+        validate_peer_endpoint_scheme(&state.config, peer_endpoint)?;
+    }
+    if state.peer_endpoint.is_none() && !state.store.is_single_replica() {
+        warn!(
+            "no gateway peer endpoint configured; this replica owns its supervisor sessions but \
+             peers cannot reach it. Single-gateway deployments are unaffected; set \
+             OPENSHELL_PEER_ENDPOINT on every replica when running more than one."
+        );
+    }
+
     if std::env::var_os("KUBERNETES_SERVICE_HOST").is_some() {
         let namespace = std::env::var("OPENSHELL_POD_NAMESPACE").ok();
         let service_account = std::env::var("OPENSHELL_SERVICE_ACCOUNT_NAME").ok();
@@ -787,20 +822,41 @@ pub(crate) async fn run_server(
                             "gateway peer ServiceAccount TokenReview authentication enabled"
                         );
                     }
+                    Err(err) if peer_routing_expected => {
+                        return Err(Error::config(format!(
+                            "in-cluster K8s client construction failed ({err}); \
+                             gateway peer authentication is required because \
+                             OPENSHELL_PEER_ENDPOINT is configured"
+                        )));
+                    }
                     Err(err) => warn!(
                         error = %err,
                         "in-cluster K8s client construction failed; \
                          gateway peer ServiceAccount authentication is disabled"
                     ),
                 }
             }
+            _ if peer_routing_expected => {
+                return Err(Error::config(
+                    "OPENSHELL_POD_NAMESPACE or OPENSHELL_SERVICE_ACCOUNT_NAME missing; \
+                     both are required for gateway peer authentication because \
+                     OPENSHELL_PEER_ENDPOINT is configured"
+                        .to_string(),
+                ));
+            }
             _ => {
                 debug!(
                     "OPENSHELL_POD_NAMESPACE or OPENSHELL_SERVICE_ACCOUNT_NAME missing; \
                      gateway peer ServiceAccount authentication disabled"
                 );
             }
         }
+    } else if peer_routing_expected {
+        return Err(Error::config(
+            "OPENSHELL_PEER_ENDPOINT is configured but the gateway is not running in a \
+             Kubernetes cluster, so gateway peer authentication is unavailable"
+                .to_string(),
+        ));
     }
 
     let state = Arc::new(state);
@@ -942,13 +998,13 @@ pub(crate) async fn run_server(
     }
 
     startup_tx.send_replace(true);
-    // The poller observes writes made by other replicas. Single-replica
-    // backends already see every write through the local update bus.
+    // The poller exists to observe writes made by other replicas. Single-
+    // replica backends have none, so it would only add load.
     if !store.is_single_replica() {
         sandbox_watch::spawn_store_poller(
             store.clone(),
             state.sandbox_watch_bus.clone(),
-            Duration::from_secs(1),
+            sandbox_watch::DEFAULT_STORE_POLL_INTERVAL,
             shutdown_rx.clone(),
         );
     }
@@ -1862,7 +1918,7 @@ mod tests {
         MultiplexService, ServerState, TlsAcceptor, allow_plaintext_service_http,
         bind_gateway_listener, classify_initial_bytes, configured_compute_driver,
         extension_token_ttl, is_benign_tls_handshake_failure, mint_gateway_extension_credential,
-        serve_gateway_listener,
+        serve_gateway_listener, validate_peer_endpoint_scheme,
     };
     use openshell_core::{
         Config,
@@ -1882,6 +1938,40 @@ mod tests {
 
     use crate::tls_test_utils::generate_test_certs_with_ca;
 
+    fn tls_enabled_config() -> Config {
+        Config::new(Some(openshell_core::TlsConfig {
+            cert_path: "/tmp/cert.pem".into(),
+            key_path: "/tmp/key.pem".into(),
+            client_ca_path: None,
+            require_client_auth: false,
+            external_cert_path: None,
+            external_key_path: None,
+            external_server_names: Vec::new(),
+        }))
+    }
+
+    #[test]
+    fn plaintext_peer_endpoint_is_rejected_on_a_tls_gateway() {
+        let error = validate_peer_endpoint_scheme(&tls_enabled_config(), "http://10.0.0.1:8080")
+            .expect_err("plaintext peer endpoint must not be accepted alongside gateway TLS");
+        assert!(
+            error.to_string().contains("plaintext"),
+            "error should name the downgrade: {error}"
+        );
+    }
+
+    #[test]
+    fn https_peer_endpoint_is_accepted_on_a_tls_gateway() {
+        validate_peer_endpoint_scheme(&tls_enabled_config(), "https://10.0.0.1:8080").unwrap();
+    }
+
+    #[test]
+    fn plaintext_peer_endpoint_is_allowed_on_a_plaintext_gateway() {
+        let config = Config::new(None);
+        assert!(config.tls.is_none());
+        validate_peer_endpoint_scheme(&config, "http://10.0.0.1:8080").unwrap();
+    }
+
     static DETECTION_PROBE_ORDER: LazyLock<Mutex<Vec<&'static str>>> =
         LazyLock::new(|| Mutex::new(Vec::new()));
 
 
@@ -40,6 +40,11 @@ pub struct PostgresStore {
 // 64-bit advisory-lock key space.
 const CROSS_OBJECT_ADVISORY_LOCK_KEY: i64 = 0x4f50_454e_5348_4c4c;
 
+// Bounds the wait for the cross-object lock. The holder only validates and
+// writes, so a wait this long means a stuck replica; failing beats blocking
+// every sandbox and provider mutation in the fleet indefinitely.
+const CROSS_OBJECT_ADVISORY_LOCK_TIMEOUT: &str = "10s";
+
 pub(super) struct PostgresAdvisoryLockGuard {
     // `close_on_drop` is set before this guard is constructed. Closing the
     // dedicated session releases the session-level advisory lock even when a
@@ -114,6 +119,11 @@ impl PostgresStore {
     ) -> PersistenceResult<PostgresAdvisoryLockGuard> {
         let mut connection = self.pool.acquire().await.map_err(|e| map_db_error(&e))?;
         connection.close_on_drop();
+        sqlx::query("SELECT set_config('lock_timeout', $1, false)")
+            .bind(CROSS_OBJECT_ADVISORY_LOCK_TIMEOUT)
+            .execute(&mut *connection)
+            .await
+            .map_err(|e| map_db_error(&e))?;
         sqlx::query("SELECT pg_advisory_lock($1)")
             .bind(CROSS_OBJECT_ADVISORY_LOCK_KEY)
             .execute(&mut *connection)
 
@@ -13,6 +13,10 @@ use tonic::Status;
 use crate::persistence::Store;
 use openshell_core::proto::Sandbox;
 
+/// How often [`spawn_store_poller`] rechecks watched sandboxes for writes made
+/// by other gateway replicas.
+pub const DEFAULT_STORE_POLL_INTERVAL: Duration = Duration::from_secs(1);
+
 /// Broadcast bus of sandbox updates keyed by sandbox id.
 ///
 /// Producers call [`SandboxWatchBus::notify`] whenever the persisted sandbox record changes.
 
@@ -100,8 +100,14 @@ impl ServiceUpstreamPool {
         let mut inner = self.inner.lock().unwrap();
         let entries = inner.idle.get_mut(key)?;
         entries.retain(|entry| is_reusable(entry, now));
-        let index = entries.iter().position(|entry| entry.sender.is_ready())?;
-        Some(entries.swap_remove(index).sender)
+        let taken = entries
+            .iter()
+            .position(|entry| entry.sender.is_ready())
+            .map(|index| entries.swap_remove(index).sender);
+        if entries.is_empty() {
+            inner.idle.remove(key);
+        }
+        taken
     }
 
     fn put(&self, key: &str, sender: UpstreamSender) {
@@ -435,19 +441,34 @@ async fn proxy_to_endpoint(
         state.service_upstreams.take(&pool_key)
     };
 
+    let reused = pooled.is_some();
     let mut sender = match pooled {
         Some(sender) => sender,
         None => open_upstream(&state, &sandbox, &endpoint, target_port, websocket_upgrade).await?,
     };
 
     let upstream = build_upstream_request(req, target_port, websocket_upgrade)?;
-    let mut response = sender.send_request(upstream).await.map_err(|err| {
-        warn!(error = %err, "sandbox service routing: upstream HTTP request failed");
-        state.service_upstreams.evict(&pool_key);
-        let route_err = ServiceRouteError::service_unreachable();
-        emit_service_relay_failure(&endpoint, target_port, route_err.reason);
-        route_err
-    })?;
+    let replay = reused.then(|| replayable_request(&upstream)).flatten();
+    let mut response = match sender.send_request(upstream).await {
+        Ok(response) => response,
+        Err(err) => {
+            warn!(error = %err, "sandbox service routing: upstream HTTP request failed");
+            state.service_upstreams.evict(&pool_key);
+            let Some(replay) = replay else {
+                let route_err = ServiceRouteError::service_unreachable();
+                emit_service_relay_failure(&endpoint, target_port, route_err.reason);
+                return Err(route_err);
+            };
+            sender =
+                open_upstream(&state, &sandbox, &endpoint, target_port, websocket_upgrade).await?;
+            sender.send_request(replay).await.map_err(|err| {
+                warn!(error = %err, "sandbox service routing: upstream HTTP retry failed");
+                let route_err = ServiceRouteError::service_unreachable();
+                emit_service_relay_failure(&endpoint, target_port, route_err.reason);
+                route_err
+            })?
+        }
+    };
 
     if !websocket_upgrade {
         state.service_upstreams.put(&pool_key, sender);
@@ -574,6 +595,23 @@ async fn load_endpoint(
         .ok_or_else(ServiceRouteError::endpoint_not_found)
 }
 
+/// Copies a request that can be sent again on a fresh connection.
+///
+/// A pooled connection can be closed by the sandbox between the liveness check
+/// and the send. Only bodyless methods are replayable, because the original
+/// body is consumed by the failed attempt.
+fn replayable_request(request: &Request<Body>) -> Option<Request<Body>> {
+    if !matches!(*request.method(), Method::GET | Method::HEAD) {
+        return None;
+    }
+    let mut replay = Request::new(Body::empty());
+    *replay.method_mut() = request.method().clone();
+    *replay.uri_mut() = request.uri().clone();
+    *replay.version_mut() = request.version();
+    *replay.headers_mut() = request.headers().clone();
+    Some(replay)
+}
+
 fn build_upstream_request(
     req: Request<Body>,
     target_port: u16,
@@ -1403,6 +1441,44 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn take_removes_an_endpoint_left_with_no_upstreams() {
+        let pool = ServiceUpstreamPool::default();
+        let (sender, _sandbox) = test_upstream().await;
+        pool.put("ep-a|8080", sender);
+
+        assert!(pool.take("ep-a|8080").is_some());
+        assert!(
+            !pool.inner.lock().unwrap().idle.contains_key("ep-a|8080"),
+            "an emptied endpoint must not linger until the next sweep"
+        );
+    }
+
+    #[test]
+    fn only_bodyless_requests_are_replayable() {
+        for method in [Method::GET, Method::HEAD] {
+            let mut request = Request::new(Body::empty());
+            *request.method_mut() = method.clone();
+            *request.uri_mut() = "/health".parse().unwrap();
+            request
+                .headers_mut()
+                .insert(header::HOST, HeaderValue::from_static("svc"));
+
+            let replay =
+                replayable_request(&request).expect("bodyless method should be replayable");
+            assert_eq!(*replay.method(), method);
+            assert_eq!(replay.uri().path(), "/health");
+            assert_eq!(replay.headers().get(header::HOST).unwrap(), "svc");
+        }
+
+        let mut post = Request::new(Body::empty());
+        *post.method_mut() = Method::POST;
+        assert!(
+            replayable_request(&post).is_none(),
+            "a request with a body cannot be replayed"
+        );
+    }
+
     #[tokio::test]
     async fn pool_does_not_hand_out_another_endpoints_upstream() {
         let pool = ServiceUpstreamPool::default();