test(exec): cover fair duplex progress and live SDK completion

Signed-off-by: Varsha Prasad Narsing <varshaprasad96@gmail.com>

Author: varshaprasad96

architecture/gateway.md

@@ -876,6 +876,8 @@ input. The gateway sends SSH EOF while keeping the output channel open until
 command completion. Input errors terminate the operation rather than masquerading
 as normal EOF. The input and output pumps are owned by the exec operation, so
 timeout or response abandonment cannot leave a detached stdin task behind.
+The pumps share polling fairly, and request processing yields cooperatively even
+for ignored resize messages, so sustained input cannot monopolize the operation.
 
 Go and TypeScript interactive-exec helpers distinguish process exit from stream
 completion. They consume the final gRPC status before reporting success and retain

crates/openshell-server/src/grpc/interactive_exec_tests.rs

@@ -9,7 +9,8 @@ use russh::server::{Auth, ChannelOpenHandle, Handler, Msg, Session};
 use std::time::Duration;
 
 struct ExecPeer {
-    echo: bool,
+    channel: Option<russh::Channel<Msg>>,
+    echo_tx: Option<mpsc::UnboundedSender<Vec<u8>>>,
     input: Arc<std::sync::Mutex<Vec<u8>>>,
     eof: Arc<AtomicBool>,
     release_output: Arc<tokio::sync::Notify>,
@@ -33,11 +34,12 @@ impl Handler for ExecPeer {
 
     async fn channel_open_session(
         &mut self,
-        _channel: russh::Channel<Msg>,
+        channel: russh::Channel<Msg>,
         reply: ChannelOpenHandle,
         _session: &mut Session,
     ) -> Result<(), Self::Error> {
         reply.accept().await;
+        self.channel = Some(channel);
         Ok(())
     }
 
@@ -48,7 +50,27 @@ impl Handler for ExecPeer {
         session: &mut Session,
     ) -> Result<(), Self::Error> {
         session.channel_success(channel)?;
-        self.echo = data == b"duplex";
+        if data == b"duplex" {
+            let channel = self.channel.take().unwrap();
+            let release = self.release_output.clone();
+            let (echo_tx, mut echo_rx) = mpsc::unbounded_channel::<Vec<u8>>();
+            self.echo_tx = Some(echo_tx);
+            self.output_task = Some(tokio::spawn(async move {
+                let (reader, writer) = channel.split();
+                // The handler receives stdin independently of output flow
+                // control, as a real process with separate I/O pumps would.
+                drop(reader);
+                while let Some(data) = echo_rx.recv().await {
+                    writer.data_bytes(data.clone()).await.unwrap();
+                    writer.extended_data_bytes(1, data).await.unwrap();
+                }
+                release.notified().await;
+                writer.exit_status(7).await.unwrap();
+                writer.close().await.unwrap();
+            }));
+        } else {
+            self.channel.take();
+        }
         if data == b"early" {
             session.exit_status_request(channel, 0)?;
             session.close(channel)?;
@@ -61,14 +83,13 @@ impl Handler for ExecPeer {
 
     async fn data(
         &mut self,
-        channel: russh::ChannelId,
+        _channel: russh::ChannelId,
         data: &[u8],
-        session: &mut Session,
+        _session: &mut Session,
     ) -> Result<(), Self::Error> {
         self.input.lock().unwrap().extend_from_slice(data);
-        if self.echo {
-            session.data(channel, data.to_vec())?;
-            session.extended_data(channel, 1, data.to_vec())?;
+        if let Some(tx) = &self.echo_tx {
+            tx.send(data.to_vec()).unwrap();
         }
         Ok(())
     }
@@ -79,6 +100,10 @@ impl Handler for ExecPeer {
         session: &mut Session,
     ) -> Result<(), Self::Error> {
         self.eof.store(true, Ordering::SeqCst);
+        self.echo_tx.take();
+        if self.output_task.is_some() {
+            return Ok(());
+        }
         let release = self.release_output.clone();
         let handle = session.handle();
         self.output_task = Some(tokio::spawn(async move {
@@ -135,7 +160,8 @@ impl Fixture {
         let eof = Arc::new(AtomicBool::new(false));
         let release_output = Arc::new(tokio::sync::Notify::new());
         let handler = ExecPeer {
-            echo: false,
+            channel: None,
+            echo_tx: None,
             input: input.clone(),
             eof: eof.clone(),
             release_output: release_output.clone(),
@@ -228,10 +254,12 @@ async fn interactive_exec_drains_stdout_and_stderr_after_input_eof() {
 async fn interactive_exec_makes_progress_in_both_directions_before_eof() {
     const CHUNKS: usize = 128;
     const CHUNK_SIZE: usize = 64 * 1024;
+    const BATCH: usize = 4;
     let fixture = Fixture::new().await;
     let (input_tx, input_rx) = mpsc::channel(16);
     let (output_tx, mut output_rx) = mpsc::channel(2);
     let drained = tokio::sync::Notify::new();
+    let progress = std::cell::Cell::new((0, 0));
     let exec = run_interactive_exec_with_russh(
         fixture.port,
         "duplex",
@@ -243,24 +271,27 @@ async fn interactive_exec_makes_progress_in_both_directions_before_eof() {
         output_tx,
     );
     let writer = async {
-        for _ in 0..CHUNKS {
+        for chunk in 1..=CHUNKS {
             input_tx
                 .send(Ok(ExecSandboxInput {
                     payload: Some(exec_sandbox_input::Payload::Stdin(vec![b'x'; CHUNK_SIZE])),
                 }))
                 .await
                 .unwrap();
+            if chunk % BATCH == 0 {
+                drained.notified().await;
+            }
         }
         // Keep the request stream open until BOTH output streams have drained.
-        // The payload exceeds SSH windows and all bridge queues, so buffering
-        // the entire exchange cannot masquerade as concurrent progress.
-        drained.notified().await;
+        // Bound in-flight data to exercise sustained interactive traffic without
+        // saturating both ends of the fixture's SSH transport simultaneously.
         drop(input_tx);
     };
     let reader = async {
         ready(&mut output_rx).await;
         let mut stdout = 0;
         let mut stderr = 0;
+        let mut acknowledged = 0;
         while stdout < CHUNKS * CHUNK_SIZE || stderr < CHUNKS * CHUNK_SIZE {
             let bytes = match output_rx.recv().await.unwrap().unwrap().payload.unwrap() {
                 exec_sandbox_event::Payload::Stdout(s) => {
@@ -271,26 +302,76 @@ async fn interactive_exec_makes_progress_in_both_directions_before_eof() {
                     stderr += s.data.len();
                     s.data
                 }
-                event => panic!("unexpected event: {event:?}"),
+                event @ exec_sandbox_event::Payload::Exit(_) => {
+                    panic!("unexpected event: {event:?}")
+                }
             };
             assert!(bytes.iter().all(|b| *b == b'x'));
+            progress.set((stdout, stderr));
             assert!(!fixture.eof.load(Ordering::SeqCst));
+            if stdout.min(stderr) >= acknowledged + BATCH * CHUNK_SIZE {
+                acknowledged += BATCH * CHUNK_SIZE;
+                drained.notify_one();
+            }
         }
         assert_eq!(stdout, CHUNKS * CHUNK_SIZE);
         assert_eq!(stderr, CHUNKS * CHUNK_SIZE);
-        drained.notify_one();
         fixture.release_output.notify_one();
         while output_rx.recv().await.is_some() {}
     };
     let (result, (), ()) = tokio::time::timeout(Duration::from_secs(30), async {
         tokio::join!(exec, writer, reader)
     })
     .await
-    .expect("stdin and stdout/stderr must make progress without request EOF");
+    .unwrap_or_else(|_| {
+        panic!(
+            "duplex stalled: input={}, output={:?}, eof={}",
+            fixture.input.lock().unwrap().len(),
+            progress.get(),
+            fixture.eof.load(Ordering::SeqCst)
+        )
+    });
     assert_eq!(result.unwrap(), 7);
     assert_eq!(fixture.input.lock().unwrap().len(), CHUNKS * CHUNK_SIZE);
 }
 
+#[tokio::test]
+async fn interactive_exec_ready_resize_stream_does_not_starve_output() {
+    use futures::StreamExt;
+    use std::sync::atomic::AtomicUsize;
+
+    // Finite to make a regression fail rather than wedge the runtime forever.
+    // These frames have no SSH write await because this session has no PTY.
+    const FRAMES: usize = 100_000;
+    let fixture = Fixture::new().await;
+    let consumed = AtomicUsize::new(0);
+    let input = futures::stream::repeat_with(|| {
+        consumed.fetch_add(1, Ordering::SeqCst);
+        Ok(ExecSandboxInput {
+            payload: Some(exec_sandbox_input::Payload::Resize(
+                openshell_core::proto::ExecSandboxWindowResize::default(),
+            )),
+        })
+    })
+    .take(FRAMES);
+    let (output_tx, mut output_rx) = mpsc::channel(2);
+    let exec =
+        run_interactive_exec_with_russh(fixture.port, "test", input, false, false, 0, 0, output_tx);
+    let reader = async {
+        ready(&mut output_rx).await;
+        assert!(
+            consumed.load(Ordering::SeqCst) < FRAMES,
+            "output must be delivered before the continuously ready input ends"
+        );
+        drop(output_rx);
+    };
+    let (result, ()) =
+        tokio::time::timeout(Duration::from_secs(5), async { tokio::join!(exec, reader) })
+            .await
+            .unwrap();
+    assert_eq!(result.unwrap_err().code(), tonic::Code::Cancelled);
+}
+
 #[tokio::test]
 async fn interactive_exec_input_error_is_not_graceful_eof() {
     for message in [

e2e/python/test_sandbox_api.py

@@ -217,7 +217,7 @@ def requests():
                         "printf '%s' \"$input\"; printf 'drained-stderr' >&2; exit 7",
                     ],
                     tty=False,
-                    timeout_seconds=20,
+                    execution_timeout=duration_pb2.Duration(seconds=20),
                 )
             )
             yield openshell_pb2.ExecSandboxInput(stdin=b"drained-stdout")

sdk/typescript/README.md

@@ -297,6 +297,7 @@ mise run sdk:ts:lint        # Biome: lint + format check (read-only)
 mise run sdk:ts:typecheck   # tsc --noEmit
 mise run sdk:ts:test        # Vitest unit tests with an 80% line-coverage gate
 mise run sdk:ts:build       # emit dist/
+mise run e2e:sdk:ts:exec    # public interactive helper against an isolated Docker gateway
 ```
 
 Formatting and linting are handled by [Biome](https://biomejs.dev) (`biome.json`): 2-space indent, single quotes, semicolons, 120-column width. Generated `src/gen/` is excluded. `sdk:ts:lint` runs in CI as part of `sdk:ts:ci`.

sdk/typescript/e2e/interactive-exec.mjs

@@ -0,0 +1,67 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+// Run through mise run e2e:sdk:ts:exec, which supplies an isolated gateway.
+import assert from 'node:assert/strict';
+import { readFileSync } from 'node:fs';
+import { join } from 'node:path';
+import { test } from 'node:test';
+import { SandboxClient } from '../dist/index.js';
+
+test('public interactive helper drains output after input EOF and verifies completion', {
+  timeout: 300_000,
+}, async () => {
+  assert.ok(process.env.XDG_CONFIG_HOME, 'run with the Docker gateway wrapper');
+  assert.ok(process.env.OPENSHELL_GATEWAY, 'run with the Docker gateway wrapper');
+  const dir = join(process.env.XDG_CONFIG_HOME, 'openshell', 'gateways', process.env.OPENSHELL_GATEWAY);
+  const metadata = JSON.parse(readFileSync(join(dir, 'metadata.json'), 'utf8'));
+  const client = await SandboxClient.connect({
+    gateway: metadata.gateway_endpoint,
+    ...(metadata.gateway_endpoint.startsWith('https:')
+      ? {
+          caCert: readFileSync(join(dir, 'mtls', 'ca.crt')),
+          clientCert: readFileSync(join(dir, 'mtls', 'tls.crt')),
+          clientKey: readFileSync(join(dir, 'mtls', 'tls.key')),
+        }
+      : {}),
+  });
+  const name = `ts-eof-${Date.now().toString(36)}`;
+  await client.create({
+    name,
+    image: process.env.OPENSHELL_E2E_DOCKER_SANDBOX_IMAGE ?? 'ghcr.io/nvidia/openshell-community/sandboxes/base:latest',
+  });
+  try {
+    await client.waitReady(name, 180);
+    const session = await client.execInteractive(
+      name,
+      ['/bin/sh', '-c', 'input=$(cat); printf "stdout:%s" "$input"; printf "stderr:drained" >&2; exit 7'],
+      { tty: false, timeoutSecs: 30, signal: AbortSignal.timeout(45_000) },
+    );
+    try {
+      // cat cannot finish until closeInput reaches the sandbox. All command
+      // output therefore proves that request EOF left the response open.
+      session.write(Buffer.from('input-before-eof'));
+      session.closeInput();
+      session.closeInput(); // The public control remains idempotent.
+      const stdout = [];
+      const stderr = [];
+      const exits = [];
+      for await (const event of session.output) {
+        assert.equal(exits.length, 0, 'exit must be the final application event');
+        if ('type' in event) exits.push(event.exitCode);
+        else (event.stream === 'stdout' ? stdout : stderr).push(event.data);
+      }
+      assert.equal(Buffer.concat(stdout).toString(), 'stdout:input-before-eof');
+      assert.equal(Buffer.concat(stderr).toString(), 'stderr:drained');
+      assert.deepEqual(exits, [7]);
+      // A nonzero process exit is not a failed transport. done resolves only
+      // after the helper has verified the final gRPC status.
+      assert.equal(await session.done, 7);
+      assert.equal(session.exitCode, 7);
+    } finally {
+      session.cancel();
+    }
+  } finally {
+    await client.delete(name);
+  }
+});

tasks/typescript.toml

@@ -59,6 +59,11 @@ depends = ["sdk:ts:proto"]
 dir = "sdk/typescript"
 run = "npm test"
 
+["e2e:sdk:ts:exec"]
+description = "Test public TypeScript interactive exec against a Docker-backed gateway"
+depends = ["sdk:ts:build"]
+run = "e2e/with-docker-gateway.sh node --test sdk/typescript/e2e/interactive-exec.mjs"
+
 ["sdk:ts:ci"]
 description = "TypeScript SDK checks (proto lint, Biome lint, codegen, typecheck, test, build)"
 depends = [