NVIDIA · krishicks · Sep 1, 2026
@@ -583,12 +583,15 @@ fn main() -> Result<()> {
         let log_push_state = if let (Some(sandbox_id), Some(endpoint)) =
             (&args.sandbox_id, &args.openshell_endpoint)
         {
-            let (tx, handle) = openshell_supervisor_process::log_push::spawn_log_push_task(
+            let (tx, drops, handle) = openshell_supervisor_process::log_push::spawn_log_push_task(
                 endpoint.clone(),
                 sandbox_id.clone(),
             );
-            let layer =
-                openshell_supervisor_process::log_push::LogPushLayer::new(sandbox_id.clone(), tx);
+            let layer = openshell_supervisor_process::log_push::LogPushLayer::new(
+                sandbox_id.clone(),
+                tx,
+                drops,
+            );
             Some((layer, handle))
         } else {
             None

@@ -38,7 +38,7 @@ socket2 = { workspace = true }
 nix = { workspace = true }
 
 # gRPC
-tonic = { workspace = true, features = ["channel", "tls-native-roots"] }
+tonic = { workspace = true, features = ["channel", "gzip", "tls-native-roots"] }
 prost = { workspace = true }
 prost-reflect = { workspace = true }
 prost-types = { workspace = true }

@@ -4088,7 +4088,32 @@ pub(super) async fn handle_push_sandbox_logs(
         )
         .await?;
 
-        for log in batch.logs.into_iter().take(100) {
+        let sandbox_dropped = state
+            .tracing_log_bus
+            .sandbox_drop_delta(&batch.sandbox_id, batch.dropped_total);
+        if sandbox_dropped > 0 {
+            metrics::counter!("openshell_sandbox_log_push_dropped_total")
+                .increment(sandbox_dropped);
+            warn!(
+                sandbox_id = %batch.sandbox_id,
+                dropped = sandbox_dropped,
+                "sandbox dropped log lines before delivery"
+            );
+        }
+
+        let dropped = log_batch_overflow(batch.logs.len());
+        if dropped > 0 {
+            metrics::counter!("openshell_sandbox_log_ingest_dropped_total")
+                .increment(dropped as u64);
+            warn!(
+                sandbox_id = %batch.sandbox_id,
+                dropped,
+                cap = MAX_LOG_LINES_PER_BATCH,
+                "sandbox log batch exceeded the per-batch cap; lines discarded"
+            );
+        }
+
+        for log in batch.logs.into_iter().take(MAX_LOG_LINES_PER_BATCH) {
             let mut log = log;
             log.source = "sandbox".to_string();
             log.sandbox_id.clone_from(&batch.sandbox_id);
@@ -4099,6 +4124,17 @@ pub(super) async fn handle_push_sandbox_logs(
     Ok(Response::new(PushSandboxLogsResponse {}))
 }
 
+/// Maximum log lines accepted from a single `PushSandboxLogs` batch.
+///
+/// The supervisor flushes at 50 but can carry up to 200 after a reconnect, so
+/// this is headroom rather than a limit it hits in normal operation.
+const MAX_LOG_LINES_PER_BATCH: usize = 200;
+
+/// Lines a batch exceeds the per-batch cap by, and so loses.
+const fn log_batch_overflow(len: usize) -> usize {
+    len.saturating_sub(MAX_LOG_LINES_PER_BATCH)
+}
+
 async fn ensure_log_stream_sandbox_scope(
     state: &Arc<ServerState>,
     principal: &Principal,
@@ -15465,6 +15501,16 @@ mod tests {
         assert_eq!(undo_err.code(), Code::NotFound);
     }
 
+    #[test]
+    fn log_batch_overflow_counts_only_lines_past_the_cap() {
+        assert_eq!(log_batch_overflow(0), 0);
+        assert_eq!(log_batch_overflow(50), 0);
+        // The supervisor's post-reconnect flush carries up to 200, which must
+        // fit without loss.
+        assert_eq!(log_batch_overflow(200), 0);
+        assert_eq!(log_batch_overflow(250), 50);
+    }
+
     #[test]
     fn build_gateway_policy_audit_event_formats_ocsf_config_line() {
         let message = build_gateway_policy_audit_event(

@@ -271,7 +271,8 @@ impl MultiplexService {
         S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
     {
         let openshell = OpenShellServer::new(OpenShellService::new(self.state.clone()))
-            .max_decoding_message_size(MAX_GRPC_DECODE_SIZE);
+            .max_decoding_message_size(MAX_GRPC_DECODE_SIZE)
+            .accept_compressed(tonic::codec::CompressionEncoding::Gzip);
         let openshell = GatewayInterceptorGrpcService::new(
             openshell,
             self.state.gateway_interceptors.clone(),

@@ -24,6 +24,7 @@ pub struct TracingLogBus {
 struct Inner {
     per_id: HashMap<String, broadcast::Sender<SandboxStreamEvent>>,
     tails: HashMap<String, VecDeque<SandboxStreamEvent>>,
+    dropped_totals: HashMap<String, u64>,
     /// Recently removed sandbox ids, in eviction order.
     removed: VecDeque<String>,
     removed_set: HashSet<String>,
@@ -42,6 +43,7 @@ impl TracingLogBus {
             inner: Arc::new(Mutex::new(Inner {
                 per_id: HashMap::new(),
                 tails: HashMap::new(),
+                dropped_totals: HashMap::new(),
                 removed: VecDeque::new(),
                 removed_set: HashSet::new(),
             })),
@@ -81,6 +83,7 @@ impl TracingLogBus {
         let mut inner = self.inner.lock().expect("tracing bus lock poisoned");
         inner.per_id.remove(sandbox_id);
         inner.tails.remove(sandbox_id);
+        inner.dropped_totals.remove(sandbox_id);
 
         if inner.removed_set.insert(sandbox_id.to_string()) {
             inner.removed.push_back(sandbox_id.to_string());
@@ -104,6 +107,22 @@ impl TracingLogBus {
             .collect()
     }
 
+    /// Return newly reported sandbox-side drops since the previous push.
+    pub(crate) fn sandbox_drop_delta(&self, sandbox_id: &str, reported_total: u64) -> u64 {
+        let mut inner = self.inner.lock().expect("tracing bus lock poisoned");
+        if inner.removed_set.contains(sandbox_id) {
+            return 0;
+        }
+
+        let seen_total = inner
+            .dropped_totals
+            .entry(sandbox_id.to_string())
+            .or_default();
+        let delta = reported_total.saturating_sub(*seen_total);
+        *seen_total = reported_total;
+        delta
+    }
+
     /// Publish a log line from an external source (e.g., sandbox push).
     ///
     /// Injects the line into the same broadcast channel and tail buffer
@@ -311,6 +330,7 @@ mod tests {
         // Create entries via subscribe and publish
         let _rx = bus.subscribe(sandbox_id);
         bus.publish_external(make_log_event(sandbox_id, "hello"));
+        assert_eq!(bus.sandbox_drop_delta(sandbox_id, 3), 3);
 
         // Verify entries exist
         assert_eq!(bus.tail(sandbox_id, 10).len(), 1);
@@ -320,6 +340,33 @@ mod tests {
 
         // Verify entries are gone
         assert!(bus.tail(sandbox_id, 10).is_empty());
+        assert!(
+            !bus.inner
+                .lock()
+                .unwrap()
+                .dropped_totals
+                .contains_key(sandbox_id)
+        );
+    }
+
+    #[test]
+    fn sandbox_drop_totals_survive_stream_reconnects() {
+        let bus = TracingLogBus::new();
+
+        assert_eq!(bus.sandbox_drop_delta("sb-reconnect", 0), 0);
+        assert_eq!(bus.sandbox_drop_delta("sb-reconnect", 5), 5);
+        assert_eq!(
+            bus.sandbox_drop_delta("sb-reconnect", 5),
+            0,
+            "reconnecting must not recount the cumulative total"
+        );
+        assert_eq!(bus.sandbox_drop_delta("sb-reconnect", 9), 4);
+        assert_eq!(
+            bus.sandbox_drop_delta("sb-reconnect", 2),
+            0,
+            "a supervisor restart must not underflow the counter"
+        );
+        assert_eq!(bus.sandbox_drop_delta("sb-reconnect", 4), 2);
     }
 
     #[test]

@@ -26,6 +26,7 @@ rand = "0.10"
 russh = "0.62"
 serde_json = { workspace = true }
 sha2 = { workspace = true }
+prost = { workspace = true }
 tokio = { workspace = true }
 tokio-stream = { workspace = true }
 tonic = { workspace = true, features = ["channel", "tls-native-roots"] }