a3s_box_runtime/grpc/
pty.rs

1//! PTY terminal client for interactive sessions.
2
3use std::path::Path;
4use std::sync::Arc;
5
6use a3s_box_core::error::{BoxError, Result};
7use tokio::sync::Mutex;
8
9type PtyFrameReader = a3s_transport::FrameReader<tokio::io::ReadHalf<tokio::net::UnixStream>>;
10type PtyFrameWriter = a3s_transport::FrameWriter<tokio::io::WriteHalf<tokio::net::UnixStream>>;
11
12/// Client for interactive PTY sessions in the guest over Unix socket.
13///
14/// Connects to the PTY server (vsock port 4090) and provides async
15/// frame-based communication for bidirectional terminal I/O.
16/// Uses `a3s_transport::FrameReader`/`FrameWriter` for wire I/O.
17#[derive(Debug)]
18pub struct PtyClient {
19    reader: PtyFrameReader,
20    writer: PtyFrameWriter,
21}
22
23impl PtyClient {
24    /// Connect to the PTY server via Unix socket.
25    pub async fn connect(socket_path: &Path) -> Result<Self> {
26        let stream = tokio::net::UnixStream::connect(socket_path)
27            .await
28            .map_err(|e| {
29                BoxError::ExecError(format!(
30                    "Failed to connect to PTY server at {}: {}",
31                    socket_path.display(),
32                    e,
33                ))
34            })?;
35
36        let (r, w) = tokio::io::split(stream);
37        Ok(Self {
38            reader: a3s_transport::FrameReader::new(r),
39            writer: a3s_transport::FrameWriter::new(w),
40        })
41    }
42
43    /// Send a PtyRequest to start an interactive session.
44    pub async fn send_request(&mut self, req: &a3s_box_core::pty::PtyRequest) -> Result<()> {
45        let payload = serde_json::to_vec(req)
46            .map_err(|e| BoxError::ExecError(format!("Failed to serialize PtyRequest: {}", e)))?;
47        self.write_raw_frame(a3s_box_core::pty::FRAME_PTY_REQUEST, &payload)
48            .await
49    }
50
51    /// Start a streaming PTY session and return a handle for supervision.
52    pub async fn start_stream(
53        mut self,
54        req: &a3s_box_core::pty::PtyRequest,
55    ) -> Result<StreamingPty> {
56        self.send_request(req).await?;
57        Ok(StreamingPty {
58            reader: self.reader,
59            writer: Arc::new(Mutex::new(self.writer)),
60            started: std::time::Instant::now(),
61            stdout_bytes: 0,
62            done: false,
63        })
64    }
65
66    /// Send terminal data to the guest.
67    pub async fn send_data(&mut self, data: &[u8]) -> Result<()> {
68        self.write_raw_frame(a3s_box_core::pty::FRAME_PTY_DATA, data)
69            .await
70    }
71
72    /// Send a terminal resize notification.
73    pub async fn send_resize(&mut self, cols: u16, rows: u16) -> Result<()> {
74        let resize = a3s_box_core::pty::PtyResize { cols, rows };
75        let payload = serde_json::to_vec(&resize)
76            .map_err(|e| BoxError::ExecError(format!("Failed to serialize PtyResize: {}", e)))?;
77        self.write_raw_frame(a3s_box_core::pty::FRAME_PTY_RESIZE, &payload)
78            .await
79    }
80
81    /// Read the next frame from the guest.
82    ///
83    /// Returns `Ok(None)` on EOF (guest disconnected).
84    pub async fn read_frame(&mut self) -> Result<Option<(u8, Vec<u8>)>> {
85        match self.reader.read_frame().await {
86            Ok(Some(frame)) => Ok(Some((frame.frame_type as u8, frame.payload))),
87            Ok(None) => Ok(None),
88            Err(e) => Err(BoxError::ExecError(format!("PTY frame read failed: {}", e))),
89        }
90    }
91
92    /// Split the client into read and write halves for concurrent I/O.
93    pub fn into_split(
94        self,
95    ) -> (
96        a3s_transport::FrameReader<tokio::io::ReadHalf<tokio::net::UnixStream>>,
97        a3s_transport::FrameWriter<tokio::io::WriteHalf<tokio::net::UnixStream>>,
98    ) {
99        (self.reader, self.writer)
100    }
101
102    /// Write a raw PTY frame using the transport writer.
103    async fn write_raw_frame(&mut self, frame_type: u8, payload: &[u8]) -> Result<()> {
104        // PTY uses custom frame type bytes (0x01-0x05) that map to transport FrameType
105        let ft = a3s_transport::FrameType::try_from(frame_type)
106            .unwrap_or(a3s_transport::FrameType::Data);
107        let frame = a3s_transport::Frame {
108            frame_type: ft,
109            payload: payload.to_vec(),
110        };
111        self.writer
112            .write_frame(&frame)
113            .await
114            .map_err(|e| BoxError::ExecError(format!("PTY frame write failed: {}", e)))
115    }
116}
117
118/// Cloneable input side for a running PTY session.
119#[derive(Clone, Debug)]
120pub struct StreamingPtyInput {
121    writer: Arc<Mutex<PtyFrameWriter>>,
122}
123
124impl StreamingPtyInput {
125    /// Write terminal data to the PTY.
126    pub async fn write_stdin(&self, data: &[u8]) -> Result<()> {
127        let frame = a3s_transport::Frame {
128            frame_type: a3s_transport::FrameType::Control,
129            payload: data.to_vec(),
130        };
131        self.writer
132            .lock()
133            .await
134            .write_frame(&frame)
135            .await
136            .map_err(|e| BoxError::ExecError(format!("PTY stdin write failed: {}", e)))
137    }
138
139    /// Request terminal resize.
140    pub async fn resize(&self, cols: u16, rows: u16) -> Result<()> {
141        let resize = a3s_box_core::pty::PtyResize { cols, rows };
142        let payload = serde_json::to_vec(&resize)
143            .map_err(|e| BoxError::ExecError(format!("Failed to serialize PtyResize: {}", e)))?;
144        let frame = a3s_transport::Frame {
145            frame_type: a3s_transport::FrameType::Heartbeat,
146            payload,
147        };
148        self.writer
149            .lock()
150            .await
151            .write_frame(&frame)
152            .await
153            .map_err(|e| BoxError::ExecError(format!("PTY resize write failed: {}", e)))
154    }
155
156    /// Close the PTY control stream. The guest treats this as a session stop.
157    pub async fn close(&self) -> Result<()> {
158        self.writer
159            .lock()
160            .await
161            .write_frame(&a3s_transport::Frame::close())
162            .await
163            .map_err(|e| BoxError::ExecError(format!("PTY close write failed: {}", e)))
164    }
165}
166
167/// Handle for reading PTY session output and exit status.
168pub struct StreamingPty {
169    reader: PtyFrameReader,
170    writer: Arc<Mutex<PtyFrameWriter>>,
171    started: std::time::Instant,
172    stdout_bytes: u64,
173    done: bool,
174}
175
176impl StreamingPty {
177    /// Return a cloneable input handle for this PTY session.
178    pub fn input(&self) -> StreamingPtyInput {
179        StreamingPtyInput {
180            writer: self.writer.clone(),
181        }
182    }
183
184    /// Read the next PTY output or exit event.
185    pub async fn next_event(&mut self) -> Result<Option<a3s_box_core::exec::ExecEvent>> {
186        use a3s_box_core::exec::{ExecChunk, ExecEvent, ExecExit, StreamType};
187
188        if self.done {
189            return Ok(None);
190        }
191
192        let Some(frame) = self
193            .reader
194            .read_frame()
195            .await
196            .map_err(|e| BoxError::ExecError(format!("PTY frame read failed: {}", e)))?
197        else {
198            self.done = true;
199            return Ok(None);
200        };
201
202        match frame.frame_type as u8 {
203            a3s_box_core::pty::FRAME_PTY_DATA => {
204                self.stdout_bytes += frame.payload.len() as u64;
205                Ok(Some(ExecEvent::Chunk(ExecChunk {
206                    stream: StreamType::Stdout,
207                    data: frame.payload,
208                })))
209            }
210            a3s_box_core::pty::FRAME_PTY_EXIT => {
211                let exit: a3s_box_core::pty::PtyExit = serde_json::from_slice(&frame.payload)
212                    .map_err(|e| BoxError::ExecError(format!("Failed to parse PTY exit: {}", e)))?;
213                self.done = true;
214                Ok(Some(ExecEvent::Exit(ExecExit {
215                    exit_code: exit.exit_code,
216                    oom_killed: false,
217                })))
218            }
219            a3s_box_core::pty::FRAME_PTY_ERROR => {
220                let msg = String::from_utf8_lossy(&frame.payload);
221                self.done = true;
222                Err(BoxError::ExecError(format!("PTY session error: {}", msg)))
223            }
224            other => Err(BoxError::ExecError(format!(
225                "Unexpected PTY frame type in stream: 0x{other:02x}"
226            ))),
227        }
228    }
229
230    /// Request cancellation of the running PTY session.
231    pub async fn cancel(&mut self) -> Result<()> {
232        self.input().close().await
233    }
234
235    /// Whether the PTY stream has finished.
236    pub fn is_done(&self) -> bool {
237        self.done
238    }
239
240    /// Get execution metrics so far.
241    pub fn metrics(&self) -> a3s_box_core::exec::ExecMetrics {
242        a3s_box_core::exec::ExecMetrics {
243            duration_ms: self.started.elapsed().as_millis() as u64,
244            peak_memory_bytes: None,
245            stdout_bytes: self.stdout_bytes,
246            stderr_bytes: 0,
247        }
248    }
249}
250
251impl std::fmt::Debug for StreamingPty {
252    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
253        f.debug_struct("StreamingPty")
254            .field("done", &self.done)
255            .field("stdout_bytes", &self.stdout_bytes)
256            .finish()
257    }
258}
259
260#[cfg(test)]
261mod tests {
262    use super::*;
263    use tokio::io::{AsyncReadExt, AsyncWriteExt};
264    use tokio::net::UnixListener;
265
266    fn bind_test_listener(path: &Path) -> Option<UnixListener> {
267        match UnixListener::bind(path) {
268            Ok(listener) => Some(listener),
269            Err(e) if e.kind() == std::io::ErrorKind::PermissionDenied => {
270                eprintln!(
271                    "skipping Unix socket test; sandbox denied bind at {}: {}",
272                    path.display(),
273                    e
274                );
275                None
276            }
277            Err(e) => panic!("failed to bind test socket {}: {}", path.display(), e),
278        }
279    }
280
281    fn test_pty_request() -> a3s_box_core::pty::PtyRequest {
282        a3s_box_core::pty::PtyRequest {
283            cmd: vec!["/bin/sh".to_string()],
284            env: vec!["TERM=xterm".to_string()],
285            working_dir: Some("/workspace".to_string()),
286            rootfs: Some("/run/a3s/rootfs".to_string()),
287            user: Some("1000:1000".to_string()),
288            cols: 100,
289            rows: 30,
290        }
291    }
292
293    #[tokio::test]
294    async fn test_pty_client_connect_nonexistent() {
295        let result = PtyClient::connect(Path::new("/tmp/nonexistent-pty-test.sock")).await;
296        assert!(result.is_err());
297    }
298
299    #[tokio::test]
300    async fn test_pty_frame_roundtrip() {
301        let tmp = tempfile::TempDir::new().unwrap();
302        let sock_path = tmp.path().join("pty.sock");
303        let Some(listener) = bind_test_listener(&sock_path) else {
304            return;
305        };
306
307        let sock_path_clone = sock_path.clone();
308        let server = tokio::spawn(async move {
309            let (mut stream, _) = listener.accept().await.unwrap();
310            // Read a frame: [type:1][len:4][payload]
311            let mut header = [0u8; 5];
312            stream.read_exact(&mut header).await.unwrap();
313            let frame_type = header[0];
314            let len = u32::from_be_bytes([header[1], header[2], header[3], header[4]]) as usize;
315            let mut payload = vec![0u8; len];
316            if len > 0 {
317                stream.read_exact(&mut payload).await.unwrap();
318            }
319            // Echo it back
320            stream.write_all(&header).await.unwrap();
321            stream.write_all(&payload).await.unwrap();
322            (frame_type, payload)
323        });
324
325        let mut client = PtyClient::connect(&sock_path_clone).await.unwrap();
326        client.send_data(b"hello world").await.unwrap();
327
328        let frame = client.read_frame().await.unwrap().unwrap();
329        assert_eq!(frame.0, a3s_box_core::pty::FRAME_PTY_DATA);
330        assert_eq!(&frame.1[..], b"hello world");
331
332        let (server_type, server_payload) = server.await.unwrap();
333        assert_eq!(server_type, a3s_box_core::pty::FRAME_PTY_DATA);
334        assert_eq!(&server_payload[..], b"hello world");
335    }
336
337    #[tokio::test]
338    async fn test_pty_send_request_serializes_payload() {
339        let tmp = tempfile::TempDir::new().unwrap();
340        let sock_path = tmp.path().join("pty_request.sock");
341        let Some(listener) = bind_test_listener(&sock_path) else {
342            return;
343        };
344
345        let server = tokio::spawn(async move {
346            let (stream, _) = listener.accept().await.unwrap();
347            let (r, _w) = tokio::io::split(stream);
348            let mut reader = a3s_transport::FrameReader::new(r);
349            reader.read_frame().await.unwrap().unwrap()
350        });
351
352        let mut client = PtyClient::connect(&sock_path).await.unwrap();
353        let req = test_pty_request();
354        client.send_request(&req).await.unwrap();
355
356        let frame = server.await.unwrap();
357        assert_eq!(frame.frame_type as u8, a3s_box_core::pty::FRAME_PTY_REQUEST);
358        let parsed: a3s_box_core::pty::PtyRequest = serde_json::from_slice(&frame.payload).unwrap();
359        assert_eq!(parsed.cmd, req.cmd);
360        assert_eq!(parsed.env, req.env);
361        assert_eq!(parsed.working_dir, req.working_dir);
362        assert_eq!(parsed.rootfs, req.rootfs);
363        assert_eq!(parsed.user, req.user);
364        assert_eq!(parsed.cols, 100);
365        assert_eq!(parsed.rows, 30);
366    }
367
368    #[tokio::test]
369    async fn test_pty_send_resize() {
370        let tmp = tempfile::TempDir::new().unwrap();
371        let sock_path = tmp.path().join("pty_resize.sock");
372        let Some(listener) = bind_test_listener(&sock_path) else {
373            return;
374        };
375
376        tokio::spawn(async move {
377            let (mut stream, _) = listener.accept().await.unwrap();
378            let mut header = [0u8; 5];
379            stream.read_exact(&mut header).await.unwrap();
380            let frame_type = header[0];
381            let len = u32::from_be_bytes([header[1], header[2], header[3], header[4]]) as usize;
382            let mut payload = vec![0u8; len];
383            stream.read_exact(&mut payload).await.unwrap();
384
385            assert_eq!(frame_type, a3s_box_core::pty::FRAME_PTY_RESIZE);
386            let resize: a3s_box_core::pty::PtyResize = serde_json::from_slice(&payload).unwrap();
387            assert_eq!(resize.cols, 120);
388            assert_eq!(resize.rows, 40);
389        });
390
391        let mut client = PtyClient::connect(&sock_path).await.unwrap();
392        client.send_resize(120, 40).await.unwrap();
393    }
394
395    #[tokio::test]
396    async fn test_pty_stream_input_writes_stdin_and_resize_frames() {
397        let tmp = tempfile::TempDir::new().unwrap();
398        let sock_path = tmp.path().join("pty_input.sock");
399        let Some(listener) = bind_test_listener(&sock_path) else {
400            return;
401        };
402
403        let server = tokio::spawn(async move {
404            let (stream, _) = listener.accept().await.unwrap();
405            let (r, _w) = tokio::io::split(stream);
406            let mut reader = a3s_transport::FrameReader::new(r);
407
408            let request = reader.read_frame().await.unwrap().unwrap();
409            assert_eq!(
410                request.frame_type as u8,
411                a3s_box_core::pty::FRAME_PTY_REQUEST
412            );
413            let stdin = reader.read_frame().await.unwrap().unwrap();
414            let resize = reader.read_frame().await.unwrap().unwrap();
415            (stdin, resize)
416        });
417
418        let client = PtyClient::connect(&sock_path).await.unwrap();
419        let stream = client.start_stream(&test_pty_request()).await.unwrap();
420        let input = stream.input();
421        input.write_stdin(b"echo hi\n").await.unwrap();
422        input.resize(132, 43).await.unwrap();
423
424        let (stdin, resize) = server.await.unwrap();
425        assert_eq!(stdin.frame_type, a3s_transport::FrameType::Control);
426        assert_eq!(stdin.payload, b"echo hi\n");
427
428        assert_eq!(resize.frame_type, a3s_transport::FrameType::Heartbeat);
429        let resize: a3s_box_core::pty::PtyResize = serde_json::from_slice(&resize.payload).unwrap();
430        assert_eq!(resize.cols, 132);
431        assert_eq!(resize.rows, 43);
432    }
433
434    #[tokio::test]
435    async fn test_pty_read_frame_eof() {
436        let tmp = tempfile::TempDir::new().unwrap();
437        let sock_path = tmp.path().join("pty_eof.sock");
438        let Some(listener) = bind_test_listener(&sock_path) else {
439            return;
440        };
441
442        tokio::spawn(async move {
443            let (stream, _) = listener.accept().await.unwrap();
444            drop(stream); // Close immediately → EOF
445        });
446
447        let mut client = PtyClient::connect(&sock_path).await.unwrap();
448        let frame = client.read_frame().await.unwrap();
449        assert!(frame.is_none()); // EOF
450    }
451
452    #[tokio::test]
453    async fn test_pty_stream_eof_marks_done_and_metrics_count_stdout() {
454        let tmp = tempfile::TempDir::new().unwrap();
455        let sock_path = tmp.path().join("pty_stream_eof.sock");
456        let Some(listener) = bind_test_listener(&sock_path) else {
457            return;
458        };
459
460        tokio::spawn(async move {
461            let (stream, _) = listener.accept().await.unwrap();
462            let (r, w) = tokio::io::split(stream);
463            let mut reader = a3s_transport::FrameReader::new(r);
464            let mut writer = a3s_transport::FrameWriter::new(w);
465
466            let request = reader.read_frame().await.unwrap().unwrap();
467            assert_eq!(
468                request.frame_type as u8,
469                a3s_box_core::pty::FRAME_PTY_REQUEST
470            );
471            writer
472                .write_frame(&a3s_transport::Frame {
473                    frame_type: a3s_transport::FrameType::Control,
474                    payload: b"abc".to_vec(),
475                })
476                .await
477                .unwrap();
478        });
479
480        let client = PtyClient::connect(&sock_path).await.unwrap();
481        let mut stream = client.start_stream(&test_pty_request()).await.unwrap();
482        assert!(!stream.is_done());
483
484        match stream.next_event().await.unwrap().unwrap() {
485            a3s_box_core::exec::ExecEvent::Chunk(chunk) => assert_eq!(chunk.data, b"abc"),
486            other => panic!("unexpected event: {other:?}"),
487        }
488        assert_eq!(stream.metrics().stdout_bytes, 3);
489
490        assert!(stream.next_event().await.unwrap().is_none());
491        assert!(stream.is_done());
492        assert!(stream.next_event().await.unwrap().is_none());
493    }
494
495    #[tokio::test]
496    async fn test_pty_client_start_stream_reads_data_and_exit() {
497        let tmp = tempfile::TempDir::new().unwrap();
498        let sock_path = tmp.path().join("pty_stream.sock");
499        let Some(listener) = bind_test_listener(&sock_path) else {
500            return;
501        };
502
503        tokio::spawn(async move {
504            let (stream, _) = listener.accept().await.unwrap();
505            let (r, w) = tokio::io::split(stream);
506            let mut reader = a3s_transport::FrameReader::new(r);
507            let mut writer = a3s_transport::FrameWriter::new(w);
508
509            let request = reader.read_frame().await.unwrap().unwrap();
510            assert_eq!(
511                request.frame_type as u8,
512                a3s_box_core::pty::FRAME_PTY_REQUEST
513            );
514
515            writer
516                .write_frame(&a3s_transport::Frame {
517                    frame_type: a3s_transport::FrameType::Control,
518                    payload: b"tty output".to_vec(),
519                })
520                .await
521                .unwrap();
522
523            let exit = a3s_box_core::pty::PtyExit { exit_code: 9 };
524            writer
525                .write_frame(&a3s_transport::Frame {
526                    frame_type: a3s_transport::FrameType::Error,
527                    payload: serde_json::to_vec(&exit).unwrap(),
528                })
529                .await
530                .unwrap();
531        });
532
533        let client = PtyClient::connect(&sock_path).await.unwrap();
534        let req = a3s_box_core::pty::PtyRequest {
535            cmd: vec!["/bin/sh".to_string()],
536            env: vec![],
537            working_dir: None,
538            rootfs: None,
539            user: None,
540            cols: 80,
541            rows: 24,
542        };
543        let mut stream = client.start_stream(&req).await.unwrap();
544
545        match stream.next_event().await.unwrap().unwrap() {
546            a3s_box_core::exec::ExecEvent::Chunk(chunk) => {
547                assert_eq!(chunk.stream, a3s_box_core::exec::StreamType::Stdout);
548                assert_eq!(chunk.data, b"tty output");
549            }
550            other => panic!("unexpected event: {other:?}"),
551        }
552        match stream.next_event().await.unwrap().unwrap() {
553            a3s_box_core::exec::ExecEvent::Exit(exit) => assert_eq!(exit.exit_code, 9),
554            other => panic!("unexpected event: {other:?}"),
555        }
556    }
557
558    #[tokio::test]
559    async fn test_pty_stream_error_frame_marks_done() {
560        let tmp = tempfile::TempDir::new().unwrap();
561        let sock_path = tmp.path().join("pty_stream_error.sock");
562        let Some(listener) = bind_test_listener(&sock_path) else {
563            return;
564        };
565
566        tokio::spawn(async move {
567            let (stream, _) = listener.accept().await.unwrap();
568            let (r, w) = tokio::io::split(stream);
569            let mut reader = a3s_transport::FrameReader::new(r);
570            let mut writer = a3s_transport::FrameWriter::new(w);
571
572            let request = reader.read_frame().await.unwrap().unwrap();
573            assert_eq!(
574                request.frame_type as u8,
575                a3s_box_core::pty::FRAME_PTY_REQUEST
576            );
577            writer
578                .write_frame(&a3s_transport::Frame {
579                    frame_type: a3s_transport::FrameType::Close,
580                    payload: b"terminal failed".to_vec(),
581                })
582                .await
583                .unwrap();
584        });
585
586        let client = PtyClient::connect(&sock_path).await.unwrap();
587        let mut stream = client.start_stream(&test_pty_request()).await.unwrap();
588
589        let err = stream.next_event().await.unwrap_err();
590        assert!(err.to_string().contains("terminal failed"));
591        assert!(stream.is_done());
592        assert!(stream.next_event().await.unwrap().is_none());
593    }
594
595    #[tokio::test]
596    async fn test_pty_stream_unexpected_frame_type_errors_without_marking_done() {
597        let tmp = tempfile::TempDir::new().unwrap();
598        let sock_path = tmp.path().join("pty_stream_unexpected.sock");
599        let Some(listener) = bind_test_listener(&sock_path) else {
600            return;
601        };
602
603        tokio::spawn(async move {
604            let (stream, _) = listener.accept().await.unwrap();
605            let (r, w) = tokio::io::split(stream);
606            let mut reader = a3s_transport::FrameReader::new(r);
607            let mut writer = a3s_transport::FrameWriter::new(w);
608
609            let request = reader.read_frame().await.unwrap().unwrap();
610            assert_eq!(
611                request.frame_type as u8,
612                a3s_box_core::pty::FRAME_PTY_REQUEST
613            );
614            writer
615                .write_frame(&a3s_transport::Frame {
616                    frame_type: a3s_transport::FrameType::Data,
617                    payload: b"not a PTY data frame".to_vec(),
618                })
619                .await
620                .unwrap();
621        });
622
623        let client = PtyClient::connect(&sock_path).await.unwrap();
624        let mut stream = client.start_stream(&test_pty_request()).await.unwrap();
625
626        let err = stream.next_event().await.unwrap_err();
627        assert!(err.to_string().contains("Unexpected PTY frame type"));
628        assert!(!stream.is_done());
629    }
630
631    #[tokio::test]
632    async fn test_pty_stream_cancel_writes_close_frame() {
633        let tmp = tempfile::TempDir::new().unwrap();
634        let sock_path = tmp.path().join("pty_cancel.sock");
635        let Some(listener) = bind_test_listener(&sock_path) else {
636            return;
637        };
638
639        tokio::spawn(async move {
640            let (stream, _) = listener.accept().await.unwrap();
641            let (r, _w) = tokio::io::split(stream);
642            let mut reader = a3s_transport::FrameReader::new(r);
643
644            let request = reader.read_frame().await.unwrap().unwrap();
645            assert_eq!(
646                request.frame_type as u8,
647                a3s_box_core::pty::FRAME_PTY_REQUEST
648            );
649            let close = reader.read_frame().await.unwrap().unwrap();
650            assert_eq!(close.frame_type, a3s_transport::FrameType::Close);
651            assert!(close.payload.is_empty());
652        });
653
654        let client = PtyClient::connect(&sock_path).await.unwrap();
655        let req = a3s_box_core::pty::PtyRequest {
656            cmd: vec!["sleep".to_string(), "60".to_string()],
657            env: vec![],
658            working_dir: None,
659            rootfs: None,
660            user: None,
661            cols: 80,
662            rows: 24,
663        };
664        let mut stream = client.start_stream(&req).await.unwrap();
665        stream.cancel().await.unwrap();
666    }
667}
a3s_box_runtime/grpc/pty.rs

a3s_box_runtime/grpc/
pty.rs
