fix(middleware): address HTTP response review findings

Signed-off-by: Piotr Mlocek <pmlocek@nvidia.com>

architecture/sandbox.md

@@ -199,7 +199,12 @@ can change. Whole-body stages delay commitment and share one non-resetting,
 supervisor-wide accumulation deadline. Body stages receive a final body result
 and then one trailer exchange; trailer mutations can only change or remove
 existing, non-protected names. Intentional blocks return the canonical 403
-before commitment and abort delivery without injected bytes after commitment.
+before commitment and abort delivery without injected bytes after commitment. Streaming input units flush after bounded coalescing even within a
+content-length body or transfer chunk. Coalescing cancels only input acquisition;
+deadline transitions and downstream writes finish outside those timeouts.
+The response runtime caps aggregate retained body data across stages and pending
+output at 8 MiB. A transformation that exceeds the budget follows its stage's
+failure policy, preserving its input when failing open.
 
 The supervisor installs policy and middleware registry changes as one runtime
 generation and preserves the last-known-good generation if preparation fails.

crates/openshell-supervisor-middleware/src/lib.rs

@@ -11,7 +11,8 @@ mod websocket;
 pub use response::{
     HttpResponseDiagnostics, HttpResponseFinish, HttpResponseInvocation,
     HttpResponseInvocationOutcome, HttpResponseMiddlewareFailure, HttpResponsePreflightInput,
-    HttpResponsePreflightOutcome, HttpResponseSession, MAX_HTTP_RESPONSE_STREAM_UNIT_BYTES,
+    HttpResponsePreflightOutcome, HttpResponseSession, MAX_HTTP_RESPONSE_RETAINED_BODY_BYTES,
+    MAX_HTTP_RESPONSE_STREAM_UNIT_BYTES,
 };
 
 pub use websocket::{

crates/openshell-supervisor-middleware/src/response.rs

@@ -32,6 +32,9 @@ use super::{
 
 const STREAM_CHANNEL_CAPACITY: usize = 4;
 pub const MAX_HTTP_RESPONSE_STREAM_UNIT_BYTES: usize = 64 * 1024;
+/// Maximum logical body bytes retained across a session's stage buffers and
+/// pending output. Temporary exchange copies have the per-binding payload cap.
+pub const MAX_HTTP_RESPONSE_RETAINED_BODY_BYTES: usize = 8 * 1024 * 1024;
 
 #[derive(Debug, Clone)]
 pub struct HttpResponsePreflightInput {
@@ -179,6 +182,7 @@ pub struct HttpResponseSession {
     invocations: Vec<HttpResponseInvocation>,
     session_admission: Option<MiddlewareSessionPermit>,
     body_transformed: bool,
+    retained_body_bytes: usize,
     defer_output_until_finish: bool,
     deferred_output: Vec<Vec<u8>>,
     connection_nominated_headers: Vec<String>,
@@ -242,6 +246,7 @@ impl HttpResponseSession {
             }
         }
         self.defer_output_until_finish = false;
+        self.release_body_bytes(&released);
         Ok(released)
     }
 
@@ -285,8 +290,16 @@ impl HttpResponseSession {
                 diagnostics: HttpResponseDiagnostics::default(),
             })?;
         let deadline = Instant::now() + MAX_MIDDLEWARE_CHAIN_TIMEOUT;
+        // Between pushes, the first active whole-body barrier owns all input
+        // not returned to the relay (at most the 4 MiB binding cap). Later
+        // barriers cannot receive bytes until it finishes or disables itself;
+        // finish consumes the session and expiry disables all such barriers.
+        // Replacement admission reserves an additional upstream unit below.
+        self.retained_body_bytes += data.len();
+        debug_assert!(self.retained_body_bytes <= MAX_HTTP_RESPONSE_RETAINED_BODY_BYTES);
         let output = self.process_units_from(0, vec![data], deadline).await?;
         if !self.defer_output_until_finish {
+            self.release_body_bytes(&output);
             return Ok(output);
         }
         if self.requires_whole_body() {
@@ -297,6 +310,7 @@ impl HttpResponseSession {
         self.defer_output_until_finish = false;
         let mut released = std::mem::take(&mut self.deferred_output);
         released.extend(output);
+        self.release_body_bytes(&released);
         Ok(released)
     }
 
@@ -369,6 +383,10 @@ impl HttpResponseSession {
         self.end_all(reason).await;
     }
 
+    fn release_body_bytes(&mut self, units: &[Vec<u8>]) {
+        self.retained_body_bytes -= units.iter().map(Vec::len).sum::<usize>();
+    }
+
     async fn process_units_from(
         &mut self,
         start: usize,
@@ -526,6 +544,31 @@ impl HttpResponseSession {
             }
         };
         let input_size = original.len();
+        let replacement_size = match &decision.action {
+            BodyAction::Transform(replacement)
+            | BodyAction::SkipRemaining(CurrentBodyAction::Transform(replacement)) => {
+                Some(replacement.len())
+            }
+            _ => None,
+        };
+        if let Some(replacement_size) = replacement_size {
+            let retained = self.retained_body_bytes - input_size + replacement_size;
+            // Reserve room for one more normalized upstream unit. Whole-body
+            // barriers bound the input retained between calls to push_body.
+            if retained
+                > MAX_HTTP_RESPONSE_RETAINED_BODY_BYTES - MAX_HTTP_RESPONSE_STREAM_UNIT_BYTES
+            {
+                return self
+                    .handle_stage_failure(
+                        index,
+                        "response_body_aggregate_over_capacity",
+                        Some(sequence),
+                        original,
+                    )
+                    .await;
+            }
+            self.retained_body_bytes = retained;
+        }
         let stage = &mut self.stages[index];
         collect_diagnostics(
             stage,
@@ -797,11 +840,11 @@ impl ChainRunner {
         entries: &[ChainEntry],
         input: HttpResponsePreflightInput,
     ) -> miette::Result<HttpResponsePreflightOutcome> {
-        validate_preflight_input(&input)?;
         let described = self.describe_http_response_chain(entries).await?;
         if described.is_empty() {
             return Ok(empty_preflight_outcome(input.headers));
         }
+        validate_preflight_input(&input)?;
         let session_admission = match self.try_reserve_middleware_session() {
             MiddlewareSessionAdmission::Admitted(admission) => admission,
             MiddlewareSessionAdmission::AtCapacity => {
@@ -852,16 +895,16 @@ impl ChainRunner {
             };
             let timeout = entry.timeout;
             let opened = tokio::time::timeout(timeout, async {
-                let mut responses = service
-                    .service
-                    .open_http_response_pre_return(receiver)
-                    .await?;
                 sender
                     .send(HttpResponseEvent {
                         event: Some(http_response_event::Event::Preflight(preflight)),
                     })
                     .await
                     .map_err(|_| tonic::Status::unavailable("middleware request stream closed"))?;
+                let mut responses = service
+                    .service
+                    .open_http_response_pre_return(receiver)
+                    .await?;
                 let response = responses.next().await.ok_or_else(|| {
                     tonic::Status::unavailable("middleware result stream closed")
                 })??;
@@ -1163,6 +1206,7 @@ impl ChainRunner {
                 invocations: Vec::new(),
                 session_admission: Some(session_admission),
                 body_transformed: false,
+                retained_body_bytes: 0,
                 defer_output_until_finish,
                 deferred_output: Vec::new(),
                 connection_nominated_headers: input.connection_nominated_headers,
@@ -1811,6 +1855,9 @@ mod tests {
         Configured,
         HangBody,
         LargeStream,
+        Expansion,
+        DeleteBody,
+        SkipBody,
         Skip,
         InvalidSkipReason,
         TrailerMutation,
@@ -1883,6 +1930,13 @@ mod tests {
             request: tonic::Request<tonic::Streaming<HttpResponseEvent>>,
         ) -> Result<tonic::Response<Self::EvaluateStream>, tonic::Status> {
             let mut requests = request.into_inner();
+            // Exercise servers that inspect the initial request before sending
+            // response headers, rather than returning a stream immediately.
+            let first = requests.next().await.expect("initial request");
+            assert!(matches!(&first, Ok(HttpResponseEvent {
+                event: Some(http_response_event::Event::Preflight(_))
+            })));
+            let mut requests = futures::stream::iter([first]).chain(requests);
             let (sender, receiver) = mpsc::channel(4);
             tokio::spawn(async move {
                 while let Some(Ok(event)) = requests.next().await {
@@ -1932,7 +1986,10 @@ mod tests {
                     operation: openshell_core::proto::SupervisorMiddlewareOperation::HttpResponse
                         as i32,
                     phase: openshell_core::proto::SupervisorMiddlewarePhase::PreReturn as i32,
-                    max_payload_bytes: if matches!(self.script, Script::LargeStream) {
+                    max_payload_bytes: if matches!(
+                        self.script,
+                        Script::LargeStream | Script::Expansion
+                    ) {
                         128 * 1024
                     } else {
                         4096
@@ -2033,6 +2090,9 @@ mod tests {
                                     | Script::InvalidSequence
                                     | Script::HangBody
                                     | Script::LargeStream
+                                    | Script::Expansion
+                                    | Script::DeleteBody
+                                    | Script::SkipBody
                                     | Script::TrailerMutation
                                     | Script::InvalidTrailerMutation => {
                                         (HttpResponseBodyMode::StreamBytes, Vec::new())
@@ -2072,6 +2132,9 @@ mod tests {
                                 break;
                             };
                             let replacement = match selected_script {
+                                Script::Expansion => vec![b'x'; 128 * 1024],
+                                Script::DeleteBody => Vec::new(),
+                                Script::SkipBody => b"replacement".to_vec(),
                                 Script::Stream
                                 | Script::InvalidSequence
                                 | Script::LargeStream
@@ -2084,6 +2147,24 @@ mod tests {
                                 | Script::Skip
                                 | Script::InvalidSkipReason => break,
                             };
+                            let transform = HttpResponseBodyTransform {
+                                replacement: Some(http_response_body_transform::Replacement::Data(
+                                    replacement,
+                                )),
+                            };
+                            let action = if matches!(selected_script, Script::SkipBody) {
+                                http_response_body_result::Action::SkipRemaining(
+                                    openshell_core::proto::HttpResponseBodySkipRemaining {
+                                        current: Some(
+                                            http_response_body_skip_remaining::Current::Transform(
+                                                transform,
+                                            ),
+                                        ),
+                                    },
+                                )
+                            } else {
+                                http_response_body_result::Action::Transform(transform)
+                            };
                             HttpResponseEventResult {
                                 result: Some(http_response_event_result::Result::BodyResult(
                                     HttpResponseBodyResult {
@@ -2095,15 +2176,7 @@ mod tests {
                                         } else {
                                             body.sequence
                                         },
-                                        action: Some(http_response_body_result::Action::Transform(
-                                            HttpResponseBodyTransform {
-                                                replacement: Some(
-                                                    http_response_body_transform::Replacement::Data(
-                                                        replacement,
-                                                    ),
-                                                ),
-                                            },
-                                        )),
+                                        action: Some(action),
                                         ..Default::default()
                                     },
                                 )),
@@ -2233,6 +2306,270 @@ mod tests {
         assert!(!modes.contains(&(HttpResponseBodyMode::StreamBytes as i32)));
     }
 
+    struct ReadPreflightBeforeOpening {
+        failure: Option<bool>,
+    }
+
+    #[tonic::async_trait]
+    impl InProcessMiddleware for ReadPreflightBeforeOpening {
+        async fn describe(&self) -> MiddlewareManifest {
+            let mut manifest = response_manifest("test/response");
+            manifest.bindings[0].timeout = "10ms".into();
+            manifest
+        }
+
+        async fn validate_config(&self, _: &str, _: &prost_types::Struct) -> miette::Result<()> {
+            Ok(())
+        }
+
+        async fn evaluate_http_request(
+            &self,
+            _: HttpRequestView<'_>,
+        ) -> miette::Result<HttpRequestResult> {
+            unreachable!()
+        }
+
+        async fn open_http_response_pre_return(
+            &self,
+            mut requests: mpsc::Receiver<HttpResponseEvent>,
+        ) -> Result<super::super::HttpResponseResultStream, tonic::Status> {
+            let first = requests.recv().await.expect("initial preflight");
+            assert!(matches!(
+                first.event,
+                Some(http_response_event::Event::Preflight(_))
+            ));
+            if let Some(hang) = self.failure {
+                if hang {
+                    futures::future::pending::<()>().await;
+                }
+                return Err(tonic::Status::unavailable("startup failed"));
+            }
+            let response = HttpResponseEventResult {
+                result: Some(http_response_event_result::Result::PreflightResult(
+                    HttpResponsePreflightResult {
+                        action: Some(http_response_preflight_result::Action::Inspect(
+                            HttpResponsePreflightInspect {
+                                body_mode: HttpResponseBodyMode::HeadersOnly as i32,
+                                header_mutations: Vec::new(),
+                            },
+                        )),
+                        ..Default::default()
+                    },
+                )),
+            };
+            Ok(Box::pin(futures::stream::iter([Ok(response)])))
+        }
+    }
+
+    #[tokio::test]
+    async fn preflight_can_be_read_before_open_returns() {
+        let runner = ChainRunner::new(Arc::new(ReadPreflightBeforeOpening { failure: None }));
+        let outcome = tokio::time::timeout(
+            Duration::from_secs(1),
+            runner.preflight_http_response(&[entry(OnError::FailClosed)], input(200)),
+        )
+        .await
+        .expect("bounded startup")
+        .expect("preflight");
+        assert!(outcome.allowed, "{}", outcome.reason);
+    }
+
+    #[tokio::test]
+    async fn preflight_opening_failure_obeys_policy_and_releases_admission() {
+        for hang in [false, true] {
+            for on_error in [OnError::FailOpen, OnError::FailClosed] {
+                let runner = ChainRunner::new(Arc::new(ReadPreflightBeforeOpening {
+                    failure: Some(hang),
+                }));
+                let permits = runner.registry.session_admission.available_permits();
+                let outcome = tokio::time::timeout(
+                    Duration::from_secs(1),
+                    runner.preflight_http_response(&[entry(on_error)], input(200)),
+                )
+                .await
+                .expect("bounded opening failure")
+                .unwrap();
+                assert_eq!(outcome.allowed, on_error == OnError::FailOpen);
+                assert!(outcome.session.is_none());
+                assert_eq!(
+                    runner.registry.session_admission.available_permits(),
+                    permits
+                );
+                assert!(outcome.invocations[0].failed);
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn multiple_whole_body_barriers_preserve_accounting_on_overflow_and_expiry() {
+        for expire in [false, true] {
+            let runner = ChainRunner::new(Arc::new(ResponseService {
+                script: Script::Configured,
+            }));
+            let mut entries = vec![
+                configured_entry("first", 0, "whole"),
+                configured_entry("second", 1, "whole"),
+                configured_entry("stream", 2, "stream"),
+            ];
+            for entry in &mut entries {
+                entry.on_error = OnError::FailOpen;
+            }
+            let mut outcome = runner
+                .preflight_http_response(&entries, input(200))
+                .await
+                .unwrap();
+            let mut session = outcome.session.take().unwrap();
+            for _ in 0..2 {
+                assert!(
+                    session
+                        .push_body(vec![b'a'; 2048])
+                        .await
+                        .unwrap()
+                        .is_empty()
+                );
+                assert!(session.retained_body_bytes <= 4096);
+                assert_eq!(
+                    session
+                        .stages
+                        .iter()
+                        .filter(|stage| !stage.whole_body.is_empty())
+                        .count(),
+                    1
+                );
+            }
+            let output = if expire {
+                session.start_whole_body_deadline(Duration::ZERO);
+                session.expire_whole_body_deadline().await.unwrap()
+            } else {
+                session.push_body(vec![b'a'; 2048]).await.unwrap()
+            };
+            assert_eq!(
+                output.concat(),
+                vec![b'A'; if expire { 4096 } else { 6144 }]
+            );
+            assert_eq!(session.retained_body_bytes, 0);
+            assert_eq!(
+                session.push_body(b"next".to_vec()).await.unwrap().concat(),
+                b"NEXT"
+            );
+            assert_eq!(session.retained_body_bytes, 0);
+            assert!(
+                session
+                    .finish(Vec::new())
+                    .await
+                    .unwrap()
+                    .body_units
+                    .is_empty()
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn deleted_and_skip_remaining_units_release_body_accounting() {
+        for script in [Script::DeleteBody, Script::SkipBody] {
+            let runner = ChainRunner::new(Arc::new(ResponseService { script }));
+            let mut outcome = runner
+                .preflight_http_response(&[entry(OnError::FailClosed)], input(200))
+                .await
+                .unwrap();
+            let mut session = outcome.session.take().unwrap();
+            for index in 0..3 {
+                let output = session.push_body(b"original".to_vec()).await.unwrap();
+                let expected = match script {
+                    Script::DeleteBody => Vec::new(),
+                    Script::SkipBody if index == 0 => b"replacement".to_vec(),
+                    Script::SkipBody => b"original".to_vec(),
+                    _ => unreachable!(),
+                };
+                assert_eq!(output.concat(), expected);
+                assert_eq!(session.retained_body_bytes, 0);
+            }
+            assert!(
+                session
+                    .finish(Vec::new())
+                    .await
+                    .unwrap()
+                    .body_units
+                    .is_empty()
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn expanding_stages_obey_aggregate_budget_and_failure_policy() {
+        for on_error in [OnError::FailClosed, OnError::FailOpen] {
+            let runner = ChainRunner::new(Arc::new(ResponseService {
+                script: Script::Expansion,
+            }));
+            let permits = runner.registry.session_admission.available_permits();
+            let entries = (0..9)
+                .map(|order| {
+                    let mut entry = entry(on_error);
+                    entry.name = format!("expand-{order}");
+                    entry.order = order;
+                    entry
+                })
+                .collect::<Vec<_>>();
+            let mut outcome = runner
+                .preflight_http_response(&entries, input(200))
+                .await
+                .unwrap();
+            let mut session = outcome.session.take().unwrap();
+            let result = tokio::time::timeout(Duration::from_secs(5), session.push_body(vec![1]))
+                .await
+                .expect("bounded expansion");
+            match result {
+                Ok(output) => {
+                    assert_eq!(on_error, OnError::FailOpen);
+                    assert!(
+                        output.iter().map(Vec::len).sum::<usize>()
+                            < MAX_HTTP_RESPONSE_RETAINED_BODY_BYTES
+                    );
+                    assert!(output.iter().flatten().all(|byte| *byte == b'x'));
+                    assert_eq!(session.retained_body_bytes, 0);
+                    assert!(
+                        session
+                            .invocations
+                            .iter()
+                            .any(|invocation| invocation.outcome
+                                == HttpResponseInvocationOutcome::FailOpen)
+                    );
+                    // Holding returned output applies backpressure: subsequent
+                    // stage work starts only when the relay calls again.
+                    drop(output);
+                    for _ in 0..3 {
+                        let output = session.push_body(vec![1]).await.unwrap();
+                        assert!(
+                            output.iter().map(Vec::len).sum::<usize>()
+                                < MAX_HTTP_RESPONSE_RETAINED_BODY_BYTES
+                        );
+                        assert_eq!(session.retained_body_bytes, 0);
+                    }
+                    let finish = session.finish(Vec::new()).await.unwrap();
+                    assert!(
+                        finish.body_units.iter().map(Vec::len).sum::<usize>()
+                            < MAX_HTTP_RESPONSE_RETAINED_BODY_BYTES
+                    );
+                }
+                Err(failure) => {
+                    assert_eq!(on_error, OnError::FailClosed);
+                    assert!(
+                        failure
+                            .reason
+                            .contains("response_body_aggregate_over_capacity")
+                    );
+                    session
+                        .end(MiddlewareSessionEndReason::MiddlewareFailure)
+                        .await;
+                }
+            }
+            assert_eq!(
+                runner.registry.session_admission.available_permits(),
+                permits
+            );
+        }
+    }
+
     #[tokio::test]
     async fn headers_only_preflight_applies_end_to_end_mutation() {
         let runner = ChainRunner::new(Arc::new(ResponseService {
@@ -2651,8 +2988,9 @@ mod tests {
         assert!(outcome.session.is_none());
 
         let _ = shutdown_tx.send(());
-        server_task
+        tokio::time::timeout(Duration::from_secs(2), server_task)
             .await
+            .expect("bounded server shutdown")
             .expect("join response middleware server")
             .expect("serve response middleware");
     }

crates/openshell-supervisor-network/src/l7/rest.rs

@@ -3378,7 +3378,21 @@ where
         guard.ensure_current()?;
     }
     let header_bytes = &buffered[..header_end];
-    let parsed = match parse_response_head_for_middleware(header_bytes) {
+    // Ordinary responses retain the HTTP parser's byte-preserving behavior.
+    // Response-specific normalization and limits apply only to selected hooks.
+    if middleware.chain.is_empty() {
+        return Ok(None);
+    }
+    let parsed = match middleware
+        .runner
+        .describe_http_response_chain(middleware.chain)
+        .await
+    {
+        Ok(described) if described.is_empty() => return Ok(None),
+        Ok(_) => parse_response_head_for_middleware(header_bytes),
+        Err(error) => Err(error),
+    };
+    let parsed = match parsed {
         Ok(parsed) => parsed,
         Err(error) => {
             debug!(error = %error, "HTTP response head normalization failed");
@@ -4412,9 +4426,10 @@ where
                 let length = usize::try_from(remaining)
                     .unwrap_or(unit_limit)
                     .min(unit_limit);
-                let unit = read_exact_response_with_deadline(
+                let unit = read_response_payload_with_deadline(
                     reader,
                     length,
+                    !pending.is_empty(),
                     session,
                     client,
                     &mut framing,
@@ -4423,6 +4438,7 @@ where
                 if let Some(guard) = generation_guard {
                     guard.ensure_current()?;
                 }
+                let partial = unit.len() < length;
                 remaining -= unit.len() as u64;
                 buffer_normalized_response_bytes(
                     session,
@@ -4433,6 +4449,15 @@ where
                     unit_limit,
                 )
                 .await?;
+                if partial {
+                    flush_normalized_response_bytes(
+                        session,
+                        client,
+                        std::mem::take(&mut pending),
+                        &mut framing,
+                    )
+                    .await?;
+                }
             }
             flush_normalized_response_bytes(session, client, pending, &mut framing).await?;
             Ok(Vec::new())
@@ -4464,9 +4489,10 @@ where
                 let mut remaining = chunk_size;
                 while remaining > 0 {
                     let length = remaining.min(unit_limit);
-                    let unit = read_exact_response_with_deadline(
+                    let unit = read_response_payload_with_deadline(
                         reader,
                         length,
+                        !pending.is_empty(),
                         session,
                         client,
                         &mut framing,
@@ -4475,6 +4501,7 @@ where
                     if let Some(guard) = generation_guard {
                         guard.ensure_current()?;
                     }
+                    let partial = unit.len() < length;
                     remaining -= unit.len();
                     buffer_normalized_response_bytes(
                         session,
@@ -4485,19 +4512,37 @@ where
                         unit_limit,
                     )
                     .await?;
+                    if partial {
+                        flush_normalized_response_bytes(
+                            session,
+                            client,
+                            std::mem::take(&mut pending),
+                            &mut framing,
+                        )
+                        .await?;
+                    }
                 }
-                if read_exact_response_with_deadline(reader, 2, session, client, &mut framing)
-                    .await?
-                    .as_slice()
-                    != b"\r\n"
+                let terminator = if let Ok(result) =
+                    tokio::time::timeout(RESPONSE_UNIT_COALESCE_TIMEOUT, reader.read_exact_vec(2))
+                        .await
                 {
+                    result?
+                } else {
+                    flush_normalized_response_bytes(
+                        session,
+                        client,
+                        std::mem::take(&mut pending),
+                        &mut framing,
+                    )
+                    .await?;
+                    read_exact_response_with_deadline(reader, 2, session, client, &mut framing)
+                        .await?
+                };
+                if terminator != b"\r\n" {
                     return Err(miette!("HTTP response chunk is missing its terminator"));
                 }
-                size_line = if let Ok(line) = tokio::time::timeout(
-                    RESPONSE_UNIT_COALESCE_TIMEOUT,
-                    read_response_line_with_deadline(reader, session, client, &mut framing),
-                )
-                .await
+                size_line = if let Ok(line) =
+                    tokio::time::timeout(RESPONSE_UNIT_COALESCE_TIMEOUT, reader.read_line()).await
                 {
                     line?
                 } else {
@@ -4513,28 +4558,45 @@ where
             }
         }
         BodyLength::None if server_wants_close || event_stream => loop {
-            let read =
-                read_response_with_deadline(reader, unit_limit, session, client, &mut framing);
-            let next = if pending.is_empty() && event_stream {
-                read.await?
-            } else if pending.is_empty() {
-                match tokio::time::timeout(RELAY_EOF_IDLE_TIMEOUT, read).await {
-                    Ok(result) => result?,
-                    Err(_) => None,
+            // Cancel only input acquisition, never expiry or client writes.
+            let wait = if pending.is_empty() {
+                if event_stream {
+                    None
+                } else {
+                    Some(RELAY_EOF_IDLE_TIMEOUT)
                 }
-            } else if let Ok(result) =
-                tokio::time::timeout(RESPONSE_UNIT_COALESCE_TIMEOUT, read).await
-            {
-                result?
             } else {
-                flush_normalized_response_bytes(
-                    session,
-                    client,
-                    std::mem::take(&mut pending),
-                    &mut framing,
-                )
-                .await?;
-                continue;
+                Some(RESPONSE_UNIT_COALESCE_TIMEOUT)
+            };
+            let read_deadline = wait.map(|wait| tokio::time::Instant::now() + wait);
+            let whole_deadline = session.whole_body_deadline();
+            let deadline = match (read_deadline, whole_deadline) {
+                (Some(a), Some(b)) => Some(a.min(b)),
+                (a, b) => a.or(b),
+            };
+            let next = if let Some(deadline) = deadline {
+                if let Ok(result) =
+                    tokio::time::timeout_at(deadline, reader.read_some(unit_limit)).await
+                {
+                    result?
+                } else {
+                    if whole_deadline.is_some_and(|d| d <= tokio::time::Instant::now()) {
+                        expire_whole_body_deadline(session, client, &mut framing).await?;
+                    } else if pending.is_empty() {
+                        return Ok(Vec::new());
+                    } else {
+                        flush_normalized_response_bytes(
+                            session,
+                            client,
+                            std::mem::take(&mut pending),
+                            &mut framing,
+                        )
+                        .await?;
+                    }
+                    continue;
+                }
+            } else {
+                reader.read_some(unit_limit).await?
             };
             let Some(unit) = next else {
                 flush_normalized_response_bytes(session, client, pending, &mut framing).await?;
@@ -4611,25 +4673,52 @@ async fn expire_whole_body_deadline<C: AsyncWrite + Unpin>(
     deliver_response_units(client, output, framing, session.requires_whole_body()).await
 }
 
-async fn read_response_with_deadline<R, C>(
+// Unit size is a maximum. Coalesce available payload without waiting for
+// the rest of a transfer chunk, and finish expiry outside read timeouts.
+async fn read_response_payload_with_deadline<R, C>(
     reader: &mut BufferedResponseReader<'_, R>,
     limit: usize,
+    has_pending: bool,
     session: &mut openshell_supervisor_middleware::HttpResponseSession,
     client: &mut C,
     framing: &mut ResponseOutputState<'_>,
-) -> Result<Option<Vec<u8>>>
+) -> Result<Vec<u8>>
 where
     R: AsyncRead + Unpin,
     C: AsyncWrite + Unpin,
 {
+    let mut payload = Vec::new();
+    let mut coalesce_deadline =
+        has_pending.then(|| tokio::time::Instant::now() + RESPONSE_UNIT_COALESCE_TIMEOUT);
     loop {
-        let Some(deadline) = session.whole_body_deadline() else {
-            return reader.read_some(limit).await;
+        let whole_deadline = session.whole_body_deadline();
+        let deadline = match (coalesce_deadline, whole_deadline) {
+            (Some(a), Some(b)) => Some(a.min(b)),
+            (a, b) => a.or(b),
         };
-        match tokio::time::timeout_at(deadline, reader.read_some(limit)).await {
-            Ok(result) => return result,
-            Err(_) => expire_whole_body_deadline(session, client, framing).await?,
+        let read = reader.read_some(limit - payload.len());
+        let result = if let Some(deadline) = deadline {
+            if let Ok(result) = tokio::time::timeout_at(deadline, read).await {
+                result
+            } else {
+                if whole_deadline.is_some_and(|d| d <= tokio::time::Instant::now()) {
+                    expire_whole_body_deadline(session, client, framing).await?;
+                }
+                if coalesce_deadline.is_some_and(|d| d <= tokio::time::Instant::now()) {
+                    return Ok(payload);
+                }
+                continue;
+            }
+        } else {
+            read.await
+        };
+        let data = result?.ok_or_else(|| miette!("HTTP response body ended unexpectedly"))?;
+        payload.extend_from_slice(&data);
+        if payload.len() == limit {
+            return Ok(payload);
         }
+        coalesce_deadline
+            .get_or_insert_with(|| tokio::time::Instant::now() + RESPONSE_UNIT_COALESCE_TIMEOUT);
     }
 }
 
@@ -5253,12 +5342,14 @@ mod tests {
         BlockWholeBody,
         BlockStream,
         SlowWholeBody,
+        SlowStream,
         InvalidBodySequence,
         InvalidWholeBodySequence,
     }
 
     struct ResponseRelayService {
         script: ResponseRelayScript,
+        request_only: bool,
     }
 
     #[tonic::async_trait]
@@ -5268,8 +5359,16 @@ mod tests {
                 name: "test/response-relay".into(),
                 service_version: "test".into(),
                 bindings: vec![MiddlewareBinding {
-                    operation: SupervisorMiddlewareOperation::HttpResponse as i32,
-                    phase: SupervisorMiddlewarePhase::PreReturn as i32,
+                    operation: if self.request_only {
+                        SupervisorMiddlewareOperation::HttpRequest
+                    } else {
+                        SupervisorMiddlewareOperation::HttpResponse
+                    } as i32,
+                    phase: if self.request_only {
+                        SupervisorMiddlewarePhase::PreCredentials
+                    } else {
+                        SupervisorMiddlewarePhase::PreReturn
+                    } as i32,
                     max_payload_bytes: 4096,
                     timeout: String::new(),
                 }],
@@ -5302,7 +5401,11 @@ mod tests {
             openshell_supervisor_middleware::HttpResponseResultStream,
             tonic::Status,
         > {
-            let script = self.script;
+            assert!(
+                !self.request_only,
+                "request-only service received a response"
+            );
+            let mut script = self.script;
             let (sender, receiver) = mpsc::channel(4);
             tokio::spawn(async move {
                 while let Some(event) = requests.recv().await {
@@ -5310,7 +5413,14 @@ mod tests {
                         break;
                     };
                     let result = match event {
-                        http_response_event::Event::Preflight(_) => {
+                        http_response_event::Event::Preflight(preflight) => {
+                            if preflight
+                                .config
+                                .as_ref()
+                                .is_some_and(|config| config.fields.contains_key("whole_body"))
+                            {
+                                script = ResponseRelayScript::WholeBody;
+                            }
                             if matches!(script, ResponseRelayScript::BlockPreflight) {
                                 HttpResponseEventResult {
                                     result: Some(
@@ -5345,6 +5455,7 @@ mod tests {
                                         (HttpResponseBodyMode::WholeBodyBytes, Vec::new())
                                     }
                                     ResponseRelayScript::Stream
+                                    | ResponseRelayScript::SlowStream
                                     | ResponseRelayScript::BlockStream
                                     | ResponseRelayScript::InvalidBodySequence => {
                                         (HttpResponseBodyMode::StreamBytes, Vec::new())
@@ -5384,6 +5495,7 @@ mod tests {
                                     [b"whole:".as_slice(), &data].concat()
                                 }
                                 ResponseRelayScript::Stream
+                                | ResponseRelayScript::SlowStream
                                 | ResponseRelayScript::BlockStream
                                 | ResponseRelayScript::InvalidBodySequence => {
                                     data.to_ascii_uppercase()
@@ -5391,7 +5503,11 @@ mod tests {
                                 ResponseRelayScript::HeadersOnly
                                 | ResponseRelayScript::BlockPreflight => break,
                             };
-                            if matches!(script, ResponseRelayScript::SlowWholeBody) {
+                            if matches!(
+                                script,
+                                ResponseRelayScript::SlowWholeBody
+                                    | ResponseRelayScript::SlowStream
+                            ) {
                                 tokio::time::sleep(std::time::Duration::from_millis(75)).await;
                             }
                             HttpResponseEventResult {
@@ -7326,6 +7442,132 @@ mod tests {
         assert!(!is_bodiless_response("POST", 201));
     }
 
+    #[tokio::test]
+    async fn response_middleware_unbound_chains_preserve_ordinary_headers() {
+        let mut many_headers = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n".to_vec();
+        for _ in 0..129 {
+            many_headers.extend_from_slice(b"Set-Cookie: a=b\r\n");
+        }
+        many_headers.extend_from_slice(b"\r\n");
+        let opaque_headers =
+            b"HTTP/1.1 200 OK\r\nContent-Length: 3\r\nX-Opaque: \xff\xfe\r\n\r\nabc".to_vec();
+        let (_, chain) = response_middleware_fixture(ResponseRelayScript::HeadersOnly);
+        let request_only_runner =
+            openshell_supervisor_middleware::ChainRunner::new(Arc::new(ResponseRelayService {
+                script: ResponseRelayScript::HeadersOnly,
+                request_only: true,
+            }));
+        let empty_runner = openshell_supervisor_middleware::ChainRunner::default();
+        for response in [many_headers, opaque_headers] {
+            assert!(response.len() < MAX_HEADER_BYTES);
+            for context in [
+                None,
+                Some(response_middleware_context(&empty_runner, &[], "GET")),
+                Some(response_middleware_context(
+                    &request_only_runner,
+                    &chain,
+                    "GET",
+                )),
+            ] {
+                let mut upstream = response.as_slice();
+                let mut delivered = Vec::new();
+                let outcome = relay_response(
+                    "GET",
+                    &mut upstream,
+                    &mut delivered,
+                    RelayResponseOptions::default(),
+                    context,
+                )
+                .await
+                .unwrap();
+                assert!(matches!(outcome, RelayOutcome::Reusable));
+                assert_eq!(delivered, response);
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn response_middleware_selected_hook_enforces_header_limits() {
+        let mut response = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n".to_vec();
+        for _ in 0..129 {
+            response.extend_from_slice(b"Set-Cookie: a=b\r\n");
+        }
+        response.extend_from_slice(b"\r\n");
+        let (runner, chain) = response_middleware_fixture(ResponseRelayScript::HeadersOnly);
+        let mut upstream = response.as_slice();
+        let mut delivered = Vec::new();
+        let outcome = relay_response(
+            "GET",
+            &mut upstream,
+            &mut delivered,
+            RelayResponseOptions::default(),
+            Some(response_middleware_context(&runner, &chain, "GET")),
+        )
+        .await
+        .unwrap();
+        assert!(matches!(outcome, RelayOutcome::Consumed));
+        assert!(delivered.starts_with(b"HTTP/1.1 502 "));
+    }
+
+    #[tokio::test]
+    async fn response_middleware_flushes_partial_framed_payload_promptly() {
+        for chunked in [false, true] {
+            let (runner, chain) = response_middleware_fixture(ResponseRelayScript::Stream);
+            let (mut upstream_read, mut upstream_write) = tokio::io::duplex(8192);
+            // Small capacity forces the relay to complete partial downstream writes.
+            let (mut client_read, mut client_write) = tokio::io::duplex(7);
+            let head = if chunked {
+                b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nContent-Type: text/event-stream\r\n\r\n1000\r\nabc".as_slice()
+            } else {
+                b"HTTP/1.1 200 OK\r\nContent-Length: 4096\r\nContent-Type: text/event-stream\r\n\r\nabc".as_slice()
+            };
+            upstream_write.write_all(head).await.unwrap();
+            let task = tokio::spawn(async move {
+                relay_response(
+                    "GET",
+                    &mut upstream_read,
+                    &mut client_write,
+                    RelayResponseOptions::default(),
+                    Some(response_middleware_context(&runner, &chain, "GET")),
+                )
+                .await
+            });
+            let result = tokio::time::timeout(std::time::Duration::from_secs(2), async {
+                let mut head = Vec::new();
+                while !head.ends_with(b"\r\n\r\n") {
+                    head.push(client_read.read_u8().await.unwrap());
+                }
+                let mut first = [0; 8];
+                client_read.read_exact(&mut first).await.unwrap();
+                assert_eq!(&first, b"3\r\nABC\r\n");
+                // Complete the same framed payload only after its first transformed
+                // bytes have reached the consumer.
+                upstream_write.write_all(&vec![b'd'; 4093]).await.unwrap();
+                if chunked {
+                    for fragment in [b"\r".as_slice(), b"\n0\r", b"\n\r", b"\n"] {
+                        upstream_write.write_all(fragment).await.unwrap();
+                        tokio::task::yield_now().await;
+                    }
+                }
+                drop(upstream_write);
+                let mut rest = Vec::new();
+                client_read.read_to_end(&mut rest).await.unwrap();
+                assert!(rest.ends_with(b"0\r\n\r\n"));
+                let body = collect_chunked_body(&mut tokio::io::empty(), &rest, None, None)
+                    .await
+                    .unwrap();
+                assert_eq!(body, vec![b'D'; 4093]);
+            })
+            .await;
+            if result.is_err() {
+                task.abort();
+            }
+            let relay = task.await;
+            assert!(result.is_ok(), "partial payload stalled, chunked={chunked}");
+            assert!(relay.unwrap().is_ok());
+        }
+    }
+
     fn response_middleware_fixture(
         script: ResponseRelayScript,
     ) -> (
@@ -7348,6 +7590,7 @@ mod tests {
         let runner =
             openshell_supervisor_middleware::ChainRunner::new(Arc::new(ResponseRelayService {
                 script,
+                request_only: false,
             }));
         let chain = vec![openshell_supervisor_middleware::ChainEntry {
             name: "response".into(),
@@ -7668,13 +7911,117 @@ mod tests {
 
         assert!(matches!(outcome.unwrap(), RelayOutcome::Reusable));
         let delivered = String::from_utf8(delivered).unwrap();
-        assert!(
-            delivered.ends_with("5\r\nhello\r\n0\r\n\r\n"),
-            "{delivered}"
-        );
+        let (_, body) = delivered.split_once("\r\n\r\n").unwrap();
+        let decoded = collect_chunked_body(&mut tokio::io::empty(), body.as_bytes(), None, None)
+            .await
+            .unwrap();
+        assert_eq!(decoded, b"hello");
         assert!(!delivered.contains("whole:hello"), "{delivered}");
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn response_middleware_expiry_preserves_bytes_through_slow_stream_and_client() {
+        for chunked in [true, false] {
+            let (runner, mut chain) = response_middleware_fixture_with_error(
+                ResponseRelayScript::SlowStream,
+                openshell_supervisor_middleware::OnError::FailOpen,
+            );
+            let mut whole_body = chain[0].clone();
+            whole_body.name = "whole-body".into();
+            whole_body
+                .config
+                .fields
+                .insert("whole_body".into(), prost_types::Value::default());
+            chain[0].order = 1;
+            chain.insert(0, whole_body);
+            let (mut upstream_read, mut upstream_write) = tokio::io::duplex(8192);
+            // Force write_all to make partial progress before each wait.
+            let (mut client_read, mut client_write) = tokio::io::duplex(7);
+            let producer = async move {
+                let head = if chunked {
+                    b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n800\r\n".as_slice()
+                } else {
+                    b"HTTP/1.1 200 OK\r\nConnection: close\r\n\r\n".as_slice()
+                };
+                upstream_write.write_all(head).await.unwrap();
+                upstream_write.write_all(&vec![b'a'; 2048]).await.unwrap();
+                if chunked {
+                    upstream_write.write_all(b"\r\n").await.unwrap();
+                }
+                // The first coalesced unit belongs to the whole-body stage. A new
+                // partial unit starts coalescing just before its deadline.
+                tokio::time::sleep(std::time::Duration::from_millis(9)).await;
+                upstream_write
+                    .write_all(if chunked { b"1\r\nb\r\n" } else { b"b" })
+                    .await
+                    .unwrap();
+                tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+                if chunked {
+                    upstream_write.write_all(b"0\r\n\r\n").await.unwrap();
+                }
+                upstream_write.shutdown().await.unwrap();
+            };
+            let relay = async {
+                let mut context = response_middleware_context(&runner, &chain, "GET");
+                context.whole_body_timeout = std::time::Duration::from_millis(10);
+                let outcome = relay_response(
+                    "GET",
+                    &mut upstream_read,
+                    &mut client_write,
+                    RelayResponseOptions::default(),
+                    Some(context),
+                )
+                .await;
+                drop(client_write);
+                outcome
+            };
+            let consumer = async move {
+                let mut delivered = Vec::new();
+                let mut bytes = [0; 7];
+                loop {
+                    let count = client_read.read(&mut bytes).await.unwrap();
+                    if count == 0 {
+                        break;
+                    }
+                    delivered.extend_from_slice(&bytes[..count]);
+                    tokio::time::sleep(std::time::Duration::from_millis(1)).await;
+                }
+                delivered
+            };
+            let ((), outcome, delivered) = tokio::time::timeout(
+                std::time::Duration::from_secs(3),
+                Box::pin(async { tokio::join!(producer, relay, consumer) }),
+            )
+            .await
+            .expect("response relay stalled");
+            assert!(outcome.is_ok(), "{outcome:?}");
+            assert!(delivered.starts_with(b"HTTP/1.1 200 OK\r\n"));
+            let head_end = delivered
+                .windows(4)
+                .position(|bytes| bytes == b"\r\n\r\n")
+                .unwrap()
+                + 4;
+            let mut wire = &delivered[head_end..];
+            let mut body = Vec::new();
+            loop {
+                let end = wire.windows(2).position(|bytes| bytes == b"\r\n").unwrap();
+                let size =
+                    usize::from_str_radix(std::str::from_utf8(&wire[..end]).unwrap(), 16).unwrap();
+                wire = &wire[end + 2..];
+                if size == 0 {
+                    assert_eq!(wire, b"\r\n");
+                    break;
+                }
+                body.extend_from_slice(&wire[..size]);
+                assert_eq!(&wire[size..size + 2], b"\r\n");
+                wire = &wire[size + 2..];
+            }
+            let mut expected = vec![b'A'; 2048];
+            expected.push(b'B');
+            assert_eq!(body, expected, "chunked={chunked}");
+        }
+    }
+
     #[tokio::test]
     async fn response_middleware_streams_normalized_chunks_and_preserves_trailers() {
         let (outcome, delivered) = run_response_middleware_relay(

docs/reference/gateway-config.mdx

@@ -315,6 +315,8 @@ The gateway connects to every registered service and validates `Describe` before
 
 `timeout` is the operator-configured service-wide RPC timeout. It accepts the same compact duration syntax as gateway interceptors: an integer followed by `ms` or `s`, such as `500ms` or `2s`. Values must be between `10ms` and `30s`, inclusive. Omit the field to use the 500 ms platform default. A binding may advertise a shorter `timeout` in the `Describe` manifest, but it cannot extend the operator-configured deadline; OpenShell uses the smaller value. OpenShell validates both levels before accepting the service. The operator-configured service timeout applies to `Describe` and `ValidateConfig`. The effective binding timeout applies to HTTP request evaluation, HTTP response preflight and unit exchanges, WebSocket preflight, and each WebSocket message. Accepted streaming protocols have no connection-wide RPC deadline.
 
+Response middleware also has an 8 MiB aggregate retained-body limit per response session across stage buffers and pending output. This limit is separate from each service's `max_payload_bytes` and cannot be configured. A replacement that exceeds the available budget follows the stage's `on_error` policy. Fail-open disables that stage and forwards its original input; fail-closed returns 502 before commitment or stops delivery after commitment. Streaming inputs flush after a short coalescing window without waiting for an entire upstream transfer chunk.
+
 `http_response_whole_body_timeout` is a supervisor-wide safety bound, not a middleware RPC timeout. It defaults to `120s` and accepts a positive integer followed by `ms`, `s`, or `m`. The timer starts after ordered response preflight selects a `WHOLE_BODY_BYTES` stage and before OpenShell reads the first response body byte. It does not reset as chunks arrive. It ends after the normalized body reaches end of stream and every whole-body barrier produces output. On expiry, each active whole-body stage follows its configured `on_error` policy: fail-open releases OpenShell-owned bytes through later stages, while fail-closed returns the canonical pre-commit 502 response. Changing this field requires restarting the gateway.
 
 The service `grpc_endpoint` supports plaintext `http://` and TLS `https://`. HTTPS uses the platform trust store unless `tls_ca_cert_path` names a certificate-only PEM bundle. OpenShell rejects bundles containing private keys, loads the certificates at gateway startup, and distributes only public certificates to sandbox supervisors; normal TLS hostname verification still applies. `audience` sets the exact audience for gateway-minted service tokens and defaults to `urn:openshell:extension:middleware:<name>`. After authenticated `Describe` succeeds, OpenShell treats a non-empty manifest `expected_audience` as a consistency assertion and refuses to start when it differs from the configured audience. A strict verifier may reject an incorrect audience before returning the manifest.

examples/supervisor-middleware-content-guard/README.md

@@ -15,7 +15,7 @@ This configured-literal guard applies the same case-sensitive terms to UTF-8 HTT
 
 ## Prerequisites
 
-Install `cargo`, `curl`, `jq`, `openssl`, and `uv` with Python 3 on the host before running the smoke script.
+Install `cargo`, `curl`, `jq`, `openssl`, `mise`, and `uv` with Python 3 on the host before running the smoke script. Start Docker or Podman. The supervisor image build uses the repository's Linux cross-compilation toolchain, including `cargo-zigbuild` and Zig on macOS. Install the repository's mise tools before running it.
 
 ## Run the smoke example
 
@@ -39,7 +39,15 @@ The script creates the sandbox and prints the guarded and unguarded request comm
 CONTENT_GUARD_SMOKE_HOST=192.168.1.10 ./examples/supervisor-middleware-content-guard/smoke.sh --test-suite
 ```
 
-The gateway auto-detects its compute driver. Set `CONTENT_GUARD_SMOKE_DRIVER=docker` or `CONTENT_GUARD_SMOKE_DRIVER=podman` if more than one local runtime is installed and auto-detection selects the wrong one.
+The script defaults to Docker. Set `CONTENT_GUARD_SMOKE_DRIVER=podman` to build and run with Podman instead.
+
+On Linux and macOS, the script runs `mise run docker:build:supervisor` with the selected container engine to build a Linux supervisor from the current checkout. It configures that driver's `supervisor_image` with a unique local tag, so the response checks exercise the local runtime changes. macOS host binaries are never used inside the sandbox. The local image remains available after the smoke run.
+
+Cargo's configured target directory applies to the host binaries and the Linux supervisor build. For example:
+
+```shell
+CARGO_TARGET_DIR=/tmp/content-guard-target ./examples/supervisor-middleware-content-guard/smoke.sh --test-suite
+```
 
 ## Run manually
 

examples/supervisor-middleware-content-guard/smoke.sh

@@ -25,7 +25,7 @@ Environment:
   CONTENT_GUARD_SMOKE_HOST  Non-loopback host address reachable from both the
                             gateway and sandbox containers.
   CONTENT_GUARD_SMOKE_DRIVER
-                            Optional compute driver name, such as docker or podman.
+                            Compute driver: docker (default) or podman.
 EOF
 }
 
@@ -105,7 +105,14 @@ detect_service_host() {
 }
 
 SERVICE_HOST="$(detect_service_host)"
-COMPUTE_DRIVER="${CONTENT_GUARD_SMOKE_DRIVER:-}"
+COMPUTE_DRIVER="${CONTENT_GUARD_SMOKE_DRIVER:-docker}"
+case "$COMPUTE_DRIVER" in
+  docker | podman) ;;
+  *)
+    echo "CONTENT_GUARD_SMOKE_DRIVER must be docker or podman" >&2
+    exit 1
+    ;;
+esac
 if [[ "$SERVICE_HOST" == "localhost" || "$SERVICE_HOST" == "::1" || "$SERVICE_HOST" == 127.* || "$SERVICE_HOST" == *:* ]]; then
   echo "CONTENT_GUARD_SMOKE_HOST must be a non-loopback IPv4 address: $SERVICE_HOST" >&2
   exit 1
@@ -121,6 +128,7 @@ MIDDLEWARE_LOG="$LOG_DIR/middleware.log"
 UPSTREAM_LOG="$LOG_DIR/upstream.log"
 SANDBOX_LOG="$LOG_DIR/sandbox.log"
 RUN_ID="content-guard-smoke-$$-$RANDOM"
+SUPERVISOR_IMAGE="localhost/openshell-content-guard/supervisor:$RUN_ID"
 # Sandbox names are capped at 19 characters. Use a short prefix with
 # the PID for uniqueness; keep the full RUN_ID for gateway identity.
 SANDBOX_NAME="cg-$$-$RANDOM"
@@ -228,8 +236,8 @@ allow_insecure_transport = true
 max_payload_bytes = 262144
 timeout = "500ms"
 
-[openshell.drivers.docker]
-supervisor_bin = "$ROOT/target/debug/openshell-sandbox"
+[openshell.drivers.$COMPUTE_DRIVER]
+supervisor_image = "$SUPERVISOR_IMAGE"
 EOF
 }
 
@@ -533,13 +541,21 @@ require_command curl
 require_command jq
 require_command openssl
 require_command uv
+require_command mise
 ROOT_TARGET_DIR="$(cargo_target_dir "$ROOT/Cargo.toml")"
 EXAMPLE_TARGET_DIR="$(cargo_target_dir "$EXAMPLE_DIR/Cargo.toml")"
 GATEWAY_BIN="$ROOT_TARGET_DIR/debug/openshell-gateway"
 CLI_BIN="$ROOT_TARGET_DIR/debug/openshell"
 MIDDLEWARE_BIN="$EXAMPLE_TARGET_DIR/debug/supervisor-middleware-content-guard"
 run_setup_step "building gateway" cargo build --quiet -p openshell-gateway --bin openshell-gateway
-run_setup_step "building sandbox supervisor" cargo build --quiet -p openshell-sandbox --bin openshell-sandbox
+# Always rebuild from this checkout and load into the selected runtime. Native
+# macOS binaries cannot run in Linux sandboxes; Podman also needs an image.
+# A unique tag prevents the driver from selecting an older published runtime.
+run_setup_step "building Linux sandbox supervisor image" \
+  env -u CI -u DOCKER_PLATFORM -u DOCKER_PUSH -u DOCKER_OUTPUT \
+  CONTAINER_ENGINE="$COMPUTE_DRIVER" PREBUILT_AUTO_STAGE=1 \
+  IMAGE_REGISTRY=localhost/openshell-content-guard IMAGE_TAG="$RUN_ID" \
+  mise run docker:build:supervisor
 run_setup_step "building content guard" cargo build --quiet --manifest-path "$EXAMPLE_DIR/Cargo.toml"
 run_setup_step "building CLI" cargo build --quiet -p openshell-cli --bin openshell
 generate_gateway_jwt_bundle

skills/generate-sandbox-policy/SKILL.md

@@ -214,7 +214,7 @@ Add `network_middlewares` only when the user asks to inspect, transform, redact,
 - Use `openshell/regex` without gateway registration for fixed-pattern redaction of UTF-8 HTTP request bodies or complete client-to-upstream WebSocket text messages.
 - Use an operator-owned middleware name only when it is already registered under `[[openshell.supervisor.middleware]]` and reachable from both the gateway and sandbox supervisors.
 - Confirm that a requested WebSocket implementation exposes a `WEBSOCKET_MESSAGE/PRE_CREDENTIALS` binding. `openshell/regex` exposes this binding. A host-matched HTTP-only implementation may inspect the upgrade GET but does not join the post-upgrade chain; messages pass and OpenShell emits `binding_not_selected` coverage regardless of `on_error`.
-- Confirm that requested response processing exposes `HTTP_RESPONSE/PRE_RETURN`. Response stages choose header-only, whole-body, or streaming inspection independently. Whole-body stages delay response commitment and are bounded by the gateway's supervisor-wide accumulation timeout. Intentional blocks produce a canonical 403 before commitment and abort the connection after commitment.
+- Confirm that requested response processing exposes `HTTP_RESPONSE/PRE_RETURN`. Response stages choose header-only, whole-body, or streaming inspection independently. Whole-body stages delay response commitment and are bounded by the gateway's supervisor-wide accumulation timeout. Expanding response transformations also share an 8 MiB retained-body budget per session; exhaustion follows the stage's `on_error` policy. Intentional blocks produce a canonical 403 before commitment and abort the connection after commitment.
 - WebSocket middleware runs for both `ws://` and `wss://` and receives complete client text messages only. Binary messages pass under both error modes and emit `unsupported_message_type` coverage for active stages. Upstream-to-client messages remain uninspected. Do not claim that V1 provides all-message WebSocket inspection.
 - Treat `fail_open` on WebSocket as a session-scoped bypass: if the stage stream fails, OpenShell disables it for later messages on that connection and emits a state-change finding. Prefer `fail_closed` for required redaction or authorization.
 - `on_error` governs failures after an advertised operation binding is selected. It does not apply to an unadvertised WebSocket binding or binary-message pass-through. An explicit HTTP, WebSocket preflight, or WebSocket message denial is authoritative under both `fail_open` and `fail_closed`.

tasks/scripts/stage-prebuilt-binaries.sh

@@ -171,6 +171,7 @@ build_component_for_arch() {
   local current_host_os
   local current_host_arch
   local binary_path
+  local cargo_output_dir
   local build_rustflags
 
   resolve_component "$component"
@@ -257,7 +258,8 @@ build_component_for_arch() {
     CARGO_INCREMENTAL=0 mise x -- "${cargo_env[@]}" "${cargo_subcommand[@]}" "${args[@]}"
   )
 
-  binary_path="${ROOT}/target/${target}/release/${binary}"
+  cargo_output_dir="$(cd "$ROOT" && mise x -- cargo metadata --format-version=1 --no-deps | jq -er '.target_directory')"
+  binary_path="${cargo_output_dir}/${target}/release/${binary}"
   if [[ "$component" == "gateway" ]]; then
     "$SCRIPT_DIR/verify-glibc-symbols.sh" 2.28 "$binary_path"
   elif [[ "$component" == "supervisor" ]]; then