From 7216d530cc60e2b0a5e1760bce84bb298b6dd606 Mon Sep 17 00:00:00 2001 From: Marut K Date: Tue, 1 Sep 2026 22:59:04 +0700 Subject: [PATCH] fix(exec): preserve PTY output under backpressure - forward output through the bounded runtime queue without dropping chunks - make the runner stream channel rendezvous-sized to avoid shutdown backlog - cover fast output tail preservation through the CLI integration path --- src/apps/config.rs | 4 ++-- src/apps/exec.rs | 20 +++++++++++--------- tests/integration.rs | 17 +++++++++++++++-- 3 files changed, 28 insertions(+), 13 deletions(-) diff --git a/src/apps/config.rs b/src/apps/config.rs index 61ec91c..ae53641 100644 --- a/src/apps/config.rs +++ b/src/apps/config.rs @@ -29,8 +29,8 @@ pub const PTY_DEFAULT_ROWS: u16 = 24; /// Height of the footer row in the full-screen output view. pub const OUTPUT_FOOTER_HEIGHT: u16 = 1; -/// Stream channel capacity (1000 messages). -pub const STREAM_CHANNEL_SIZE: usize = 1000; +/// Capacity of the raw PTY stream channel (`0` for rendezvous backpressure). +pub const STREAM_CHANNEL_SIZE: usize = 0; // Preview layout constants diff --git a/src/apps/exec.rs b/src/apps/exec.rs index 25d0f3d..37a8b5a 100644 --- a/src/apps/exec.rs +++ b/src/apps/exec.rs @@ -3,14 +3,13 @@ //! Owns the per-code-block output state and provides shared functions //! for running code, processing PTY output streams, and reloading documents. -use std::path::PathBuf; +use std::{collections::HashMap, path::PathBuf}; use crossbeam_channel::Receiver; use crate::apps::config::Envs; use crate::apps::task::Task; use crate::{pty::process::Size as PtySize, pty::stream::Stream, runner}; -use std::collections::HashMap; use upmd_parser::{nodes, CodeId}; use upmd_runtime::Cmd; @@ -104,19 +103,22 @@ pub fn merge_envs(dest: &mut Envs, captured: &Envs) { /// Creates a stream command that forwards process output and control separately. /// -/// PTY output can be effectively infinite (`yes` is the canonical case), so -/// `Out` is best-effort on the low-priority queue. Lifecycle/state messages go -/// to the high-priority queue so `Exit`/`End` cannot sit behind stale output. +/// PTY output uses the bounded low-priority queue, which applies backpressure +/// instead of dropping chunks. Lifecycle/state messages use the high-priority +/// queue so the runtime handles them before already-queued output. pub fn stream_rx( id: CodeId, rx: Receiver, mk_msg: impl Fn(CodeId, Stream) -> M + Send + 'static, ) -> Cmd { Cmd::priority_stream(move |output_tx, control_tx| { - while let Ok(msg) = rx.recv() { - if matches!(msg, Stream::Out(_)) { - let _ = output_tx.try_send(mk_msg(id, msg)); - } else if control_tx.send(mk_msg(id, msg)).is_err() { + while let Ok(stream) = rx.recv() { + let sender = if matches!(stream, Stream::Out(_)) { + &output_tx + } else { + &control_tx + }; + if sender.send(mk_msg(id, stream)).is_err() { break; } } diff --git a/tests/integration.rs b/tests/integration.rs index 7a956a6..7b91fee 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -8,12 +8,21 @@ use std::process::Command; #[cfg(unix)] #[test] -fn cli_block_yes_prints_pty_output_before_exit() { +fn cli_block_prints_pty_output_tail_before_exit() { let tmp = tempfile::tempdir().expect("create temp dir"); std::fs::write(tmp.path().join("listed-file.txt"), "").expect("create listed file"); let markdown = tmp.path().join("case.md"); - std::fs::write(&markdown, "```shell\nls\n```\n").expect("write markdown"); + std::fs::write( + &markdown, + r#"```shell +for i in $(seq 1 5000); do printf 'line-%04d\n' "$i"; done +ls +printf 'FINAL-PTY-TAIL\n' +``` +"#, + ) + .expect("write markdown"); let output = Command::new(env!("CARGO_BIN_EXE_upmd")) .arg(&markdown) @@ -33,6 +42,10 @@ fn cli_block_yes_prints_pty_output_before_exit() { stdout.contains("listed-file.txt"), "CLI output should include PTY stdout\nstdout:\n{stdout}\nstderr:\n{stderr}" ); + assert!( + stdout.contains("FINAL-PTY-TAIL"), + "CLI output should include the final PTY output\nstdout:\n{stdout}\nstderr:\n{stderr}" + ); } #[cfg(unix)]