deb: change ui/ux of pkh deb

src/context/api.rs

@@ -5,6 +5,25 @@ use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::sync::Mutex;
 
+/// Stream from which a captured output line originates
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Stream {
+    /// Standard output of the subprocess
+    Stdout,
+    /// Standard error of the subprocess
+    Stderr,
+}
+
+/// Receiver for lines captured from a subprocess
+///
+/// Implementations receive every decoded line (ANSI escapes stripped,
+/// carriage-return progress fragments collapsed) as soon as it is produced,
+/// allowing live UIs to display and rewrite subprocess output while it runs.
+pub trait LineSink: Send + Sync {
+    /// Called for each captured line
+    fn line(&self, stream: Stream, line: &str);
+}
+
 use super::local::LocalDriver;
 use super::schroot::SchrootDriver;
 use super::ssh::SshDriver;
@@ -22,6 +41,23 @@ pub trait ContextDriver {
         env: &[(String, String)],
         cwd: Option<&str>,
     ) -> io::Result<std::process::ExitStatus>;
+    /// Run a command, capturing its output line by line into `sink`
+    ///
+    /// Implementations should pipe the subprocess standard streams and forward
+    /// each decoded line to `sink` instead of letting it inherit the terminal.
+    /// The default implementation ignores the sink and behaves like
+    /// [`ContextDriver::run`].
+    fn run_captured(
+        &self,
+        program: &str,
+        args: &[String],
+        env: &[(String, String)],
+        cwd: Option<&str>,
+        sink: Arc<dyn LineSink>,
+    ) -> io::Result<std::process::ExitStatus> {
+        let _ = sink;
+        self.run(program, args, env, cwd)
+    }
     fn run_output(
         &self,
         program: &str,
@@ -140,6 +176,7 @@ impl Context {
             args: Vec::new(),
             env: Vec::new(),
             cwd: None,
+            sink: None,
         }
     }
 
@@ -250,6 +287,7 @@ pub struct ContextCommand<'a> {
     args: Vec<String>,
     env: Vec<(String, String)>,
     cwd: Option<String>,
+    sink: Option<Arc<dyn LineSink>>,
 }
 
 impl<'a> ContextCommand<'a> {
@@ -303,15 +341,33 @@ impl<'a> ContextCommand<'a> {
         self
     }
 
+    /// Enable line-wise capture of the command output into `sink`
+    ///
+    /// When a sink is set, [`ContextCommand::status`] pipes the subprocess
+    /// standard streams and forwards each decoded line to the sink instead of
+    /// letting the child inherit the terminal. Without a sink, behavior is
+    /// unchanged.
+    pub fn capture(&mut self, sink: Arc<dyn LineSink>) -> &mut Self {
+        self.sink = Some(sink);
+        self
+    }
+
     /// Run command and obtain exit status
     pub fn status(&mut self) -> io::Result<std::process::ExitStatus> {
         let program = self.program.clone();
-        self.context
-            .driver()
-            .as_ref()
-            .unwrap()
-            .run(&self.program, &self.args, &self.env, self.cwd.as_deref())
-            .map_err(|e| contextualize_spawn_error(&program, e))
+        let driver_guard = self.context.driver();
+        let driver = driver_guard.as_ref().unwrap();
+        let result = match &self.sink {
+            Some(sink) => driver.run_captured(
+                &self.program,
+                &self.args,
+                &self.env,
+                self.cwd.as_deref(),
+                sink.clone(),
+            ),
+            None => driver.run(&self.program, &self.args, &self.env, self.cwd.as_deref()),
+        };
+        result.map_err(|e| contextualize_spawn_error(&program, e))
     }
 
     /// Run command, capturing output

src/context/capture.rs

@@ -0,0 +1,125 @@
+//! Shared helpers for capturing subprocess output line by line
+//!
+//! Used by the context drivers implementing
+//! [`ContextDriver::run_captured`](super::api::ContextDriver::run_captured):
+//! raw bytes are read incrementally, split into lines, cleaned up (ANSI escape
+//! stripping, carriage-return progress collapsing) and forwarded to a
+//! [`LineSink`](super::api::LineSink).
+
+use std::io::Read;
+use std::sync::OnceLock;
+
+use super::api::{LineSink, Stream};
+use regex::Regex;
+
+/// Precompiled regex matching ANSI escape sequences (CSI and simple escapes)
+fn ansi_regex() -> &'static Regex {
+    static RE: OnceLock<Regex> = OnceLock::new();
+    RE.get_or_init(|| Regex::new(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])").unwrap())
+}
+
+/// Strip ANSI escape sequences from a line
+pub(crate) fn strip_ansi(line: &str) -> String {
+    ansi_regex().replace_all(line, "").to_string()
+}
+
+/// Read `reader` to EOF, forwarding each decoded line to `sink`
+///
+/// Lines are split on `\n`; when a line contains carriage returns (e.g. apt's
+/// `0% [Working]` progress fragments), only the last `\r`-segment is kept so
+/// progress updates replace each other instead of accumulating. Empty lines
+/// are dropped.
+pub(crate) fn pump<R: Read>(mut reader: R, stream: Stream, sink: &dyn LineSink) {
+    let mut buf: Vec<u8> = Vec::with_capacity(8192);
+    let mut chunk = [0u8; 4096];
+
+    loop {
+        match reader.read(&mut chunk) {
+            Ok(0) => break,
+            Ok(n) => buf.extend_from_slice(&chunk[..n]),
+            Err(_) => break,
+        }
+
+        while let Some(pos) = buf.iter().position(|&b| b == b'\n') {
+            let line: Vec<u8> = buf.drain(..=pos).collect();
+            emit(&line[..line.len() - 1], stream, sink);
+        }
+    }
+
+    // Flush a trailing line without newline, if any
+    if !buf.is_empty() {
+        emit(&buf, stream, sink);
+    }
+}
+
+/// Clean up and forward one raw line to the sink
+fn emit(raw: &[u8], stream: Stream, sink: &dyn LineSink) {
+    let mut line = String::from_utf8_lossy(raw).to_string();
+
+    // Strip a trailing carriage return left over from CRLF line endings
+    if line.ends_with('\r') {
+        line.pop();
+    }
+
+    // Carriage-return progress: keep only the last segment of the line
+    if let Some(idx) = line.rfind('\r') {
+        line = line[idx + 1..].to_string();
+    }
+
+    let line = strip_ansi(&line);
+    let trimmed = line.trim_end();
+    if trimmed.trim().is_empty() {
+        return;
+    }
+
+    sink.line(stream, trimmed);
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Mutex;
+
+    #[derive(Default)]
+    struct CollectingSink(Mutex<Vec<(Stream, String)>>);
+
+    impl LineSink for CollectingSink {
+        fn line(&self, stream: Stream, line: &str) {
+            self.0.lock().unwrap().push((stream, line.to_string()));
+        }
+    }
+
+    #[test]
+    fn test_pump_splits_lines_and_merges_cr_fragments() {
+        let sink = CollectingSink::default();
+        let data =
+            b"Get:1 http://x InRelease [1 kB]\r0% [Working]\r\nHit:2 http://y Release\npartial";
+        pump(&data[..], Stream::Stdout, &sink);
+
+        let lines = sink.0.lock().unwrap().clone();
+        assert_eq!(
+            lines,
+            vec![
+                (Stream::Stdout, "0% [Working]".to_string()),
+                (Stream::Stdout, "Hit:2 http://y Release".to_string()),
+                (Stream::Stdout, "partial".to_string()),
+            ]
+        );
+    }
+
+    #[test]
+    fn test_pump_strips_ansi_and_skips_empty_lines() {
+        let sink = CollectingSink::default();
+        let data = b"\x1b[1mSetting up foo\x1b[0m\n\n   \nE: boom\n";
+        pump(&data[..], Stream::Stderr, &sink);
+
+        let lines = sink.0.lock().unwrap().clone();
+        assert_eq!(
+            lines,
+            vec![
+                (Stream::Stderr, "Setting up foo".to_string()),
+                (Stream::Stderr, "E: boom".to_string()),
+            ]
+        );
+    }
+}

src/context/local.rs

@@ -1,10 +1,12 @@
 /// Local context: execute commands locally
 /// Context driver: Does nothing
-use super::api::ContextDriver;
+use super::api::{ContextDriver, LineSink, Stream};
+use super::capture::pump;
 use std::io;
 use std::os::unix::fs::symlink;
 use std::path::{Path, PathBuf};
-use std::process::Command;
+use std::process::{Command, Stdio};
+use std::sync::Arc;
 use std::time::SystemTime;
 
 pub struct LocalDriver;
@@ -80,6 +82,51 @@ impl ContextDriver for LocalDriver {
         cmd.status()
     }
 
+    fn run_captured(
+        &self,
+        program: &str,
+        args: &[String],
+        env: &[(String, String)],
+        cwd: Option<&str>,
+        sink: Arc<dyn LineSink>,
+    ) -> io::Result<std::process::ExitStatus> {
+        let mut cmd = Command::new(program);
+        cmd.args(args).envs(env.iter().map(|(k, v)| (k, v)));
+        // Best-effort: ask children not to emit ANSI colors; captured lines are
+        // stripped anyway.
+        cmd.env("NO_COLOR", "1");
+        if let Some(dir) = cwd {
+            cmd.current_dir(dir);
+        }
+        cmd.stdout(Stdio::piped())
+            .stderr(Stdio::piped())
+            .stdin(Stdio::null());
+
+        let mut child = cmd.spawn()?;
+        let stdout = child.stdout.take();
+        let stderr = child.stderr.take();
+
+        // One reader thread per stream; lines are forwarded to the sink as
+        // they arrive so the UI stays live.
+        let out_sink = sink.clone();
+        let t_out =
+            stdout.map(|r| std::thread::spawn(move || pump(r, Stream::Stdout, out_sink.as_ref())));
+        let err_sink = sink.clone();
+        let t_err =
+            stderr.map(|r| std::thread::spawn(move || pump(r, Stream::Stderr, err_sink.as_ref())));
+
+        let status = child.wait();
+
+        if let Some(t) = t_out {
+            let _ = t.join();
+        }
+        if let Some(t) = t_err {
+            let _ = t.join();
+        }
+
+        status
+    }
+
     fn run_output(
         &self,
         program: &str,

src/context/mod.rs

@@ -1,11 +1,12 @@
 mod api;
+mod capture;
 mod local;
 mod manager;
 mod schroot;
 mod ssh;
 mod unshare;
 
-pub use api::{Context, ContextCommand, ContextConfig};
+pub use api::{Context, ContextCommand, ContextConfig, LineSink, Stream};
 pub use manager::ContextManager;
 use std::sync::Arc;
 

src/context/schroot.rs

@@ -1,6 +1,6 @@
 /// Schroot context: execute commands in a schroot session
 /// Not tested, will need more work!
-use super::api::ContextDriver;
+use super::api::{ContextDriver, LineSink};
 use std::io;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
@@ -72,6 +72,71 @@ impl SchrootDriver {
             .trim()
             .to_string())
     }
+
+    /// Build the `schroot` argument list executing `(program, args)` inside
+    /// the given session.
+    ///
+    /// When `preserve_env` is set, `-p` is passed so the host environment is
+    /// preserved (historical behavior of `run()`, while `run_output()` does
+    /// not preserve it).
+    fn schroot_args(
+        session_id: &str,
+        preserve_env: bool,
+        program: &str,
+        args: &[String],
+        env: &[(String, String)],
+        cwd: Option<&str>,
+    ) -> Vec<String> {
+        let mut command_args = Vec::new();
+        if preserve_env {
+            command_args.push("-p".to_string());
+        }
+        command_args.extend([
+            "-r".to_string(),
+            "-c".to_string(),
+            session_id.to_string(),
+            "--".to_string(),
+        ]);
+
+        let (actual_program, actual_args) = Self::wrap_command(program, args, env, cwd);
+        command_args.push(actual_program);
+        command_args.extend(actual_args);
+        command_args
+    }
+
+    /// Wrap `(program, args)` in `sh -c` when a working directory or
+    /// environment variables are needed.
+    fn wrap_command(
+        program: &str,
+        args: &[String],
+        env: &[(String, String)],
+        cwd: Option<&str>,
+    ) -> (String, Vec<String>) {
+        let mut actual_program = program.to_string();
+        let mut actual_args = args.to_vec();
+
+        if cwd.is_some() || !env.is_empty() {
+            let mut shell_cmd = String::new();
+
+            if let Some(dir) = cwd {
+                shell_cmd.push_str(&format!("cd {} && ", dir));
+            }
+
+            if !env.is_empty() {
+                shell_cmd.push_str("env ");
+                for (k, v) in env {
+                    shell_cmd.push_str(&format!("{}={} ", k, v));
+                }
+            }
+
+            shell_cmd.push_str(&format!("{} {}", program, args.join(" ")));
+
+            actual_program = "sh".to_string();
+            actual_args = vec!["-c".to_string(), shell_cmd];
+        }
+
+        (actual_program, actual_args)
+    }
 }
 
 impl ContextDriver for SchrootDriver {
@@ -119,47 +184,26 @@ impl ContextDriver for SchrootDriver {
         cwd: Option<&str>,
     ) -> io::Result<std::process::ExitStatus> {
         let session_id = self.ensure_session()?;
+        let cmd_args = Self::schroot_args(&session_id, true, program, args, env, cwd);
+        self.parent().command("schroot").args(cmd_args).status()
+    }
 
-        // Construct the schroot command
-        // schroot -p -r -c session_id -- program args...
-        // If cwd is specified, we wrap in sh -c "cd cwd && ..."
-
-        let mut command_args = vec![
-            "-p".to_string(),
-            "-r".to_string(),
-            "-c".to_string(),
-            session_id,
-            "--".to_string(),
-        ];
-
-        let mut actual_program = program.to_string();
-        let mut actual_args = args.to_vec();
-
-        // Simplest: Wrap everything in `sh -c` if CWD or ENV is needed.
-        if cwd.is_some() || !env.is_empty() {
-            let mut shell_cmd = String::new();
-
-            if let Some(dir) = cwd {
-                shell_cmd.push_str(&format!("cd {} && ", dir));
-            }
-
-            if !env.is_empty() {
-                shell_cmd.push_str("env ");
-                for (k, v) in env {
-                    shell_cmd.push_str(&format!("{}={} ", k, v));
-                }
-            }
-
-            shell_cmd.push_str(&format!("{} {}", program, args.join(" ")));
-
-            actual_program = "sh".to_string();
-            actual_args = vec!["-c".to_string(), shell_cmd];
-        }
-
-        command_args.push(actual_program);
-        command_args.extend(actual_args);
-
-        self.parent().command("schroot").args(command_args).status()
+    fn run_captured(
+        &self,
+        program: &str,
+        args: &[String],
+        env: &[(String, String)],
+        cwd: Option<&str>,
+        sink: Arc<dyn LineSink>,
+    ) -> io::Result<std::process::ExitStatus> {
+        let session_id = self.ensure_session()?;
+        let cmd_args = Self::schroot_args(&session_id, true, program, args, env, cwd);
+        // Forward the sink to the wrapping command so capture chains through
+        // the parent context driver (e.g. schroot over ssh).
+        let parent = self.parent();
+        let mut cmd = parent.command("schroot");
+        cmd.args(cmd_args).capture(sink);
+        cmd.status()
     }
 
     fn run_output(
@@ -170,41 +214,8 @@ impl ContextDriver for SchrootDriver {
         cwd: Option<&str>,
     ) -> io::Result<std::process::Output> {
         let session_id = self.ensure_session()?;
-
-        let mut command_args = vec![
-            "-r".to_string(),
-            "-c".to_string(),
-            session_id,
-            "--".to_string(),
-        ];
-
-        let mut actual_program = program.to_string();
-        let mut actual_args = args.to_vec();
-
-        if cwd.is_some() || !env.is_empty() {
-            let mut shell_cmd = String::new();
-
-            if let Some(dir) = cwd {
-                shell_cmd.push_str(&format!("cd {} && ", dir));
-            }
-
-            if !env.is_empty() {
-                shell_cmd.push_str("env ");
-                for (k, v) in env {
-                    shell_cmd.push_str(&format!("{}={} ", k, v));
-                }
-            }
-
-            shell_cmd.push_str(&format!("{} {}", program, args.join(" ")));
-
-            actual_program = "sh".to_string();
-            actual_args = vec!["-c".to_string(), shell_cmd];
-        }
-
-        command_args.push(actual_program);
-        command_args.extend(actual_args);
-
-        self.parent().command("schroot").args(command_args).output()
+        let cmd_args = Self::schroot_args(&session_id, false, program, args, env, cwd);
+        self.parent().command("schroot").args(cmd_args).output()
     }
 
     fn create_temp_dir(&self) -> io::Result<String> {

src/context/ssh.rs

@@ -1,6 +1,7 @@
 /// SSH context: execute commands over an SSH connection
 /// Context driver: Copies over SFTP with ssh2, executes commands over ssh2 channels
-use super::api::ContextDriver;
+use super::api::{ContextDriver, LineSink, Stream};
+use super::capture::pump;
 use log::debug;
 use ssh2;
 use std::fs;
@@ -11,6 +12,7 @@ use std::net::TcpStream;
 use std::os::unix::process::ExitStatusExt;
 use std::path::{Path, PathBuf};
 use std::process::ExitStatus;
+use std::sync::Arc;
 
 pub fn connect_ssh(host: &str, user: Option<&str>, port: Option<u16>) -> io::Result<ssh2::Session> {
     let port = port.unwrap_or(22);
@@ -139,6 +141,55 @@ impl ContextDriver for SshDriver {
         Ok(ExitStatus::from_raw(code))
     }
 
+    fn run_captured(
+        &self,
+        program: &str,
+        args: &[String],
+        env: &[(String, String)],
+        cwd: Option<&str>,
+        sink: Arc<dyn LineSink>,
+    ) -> io::Result<std::process::ExitStatus> {
+        let sess = connect_ssh(&self.host, self.user.as_deref(), self.port)?;
+        let mut channel = sess.channel_session().map_err(io::Error::other)?;
+
+        // Construct command line with env vars (same escaping as `run`)
+        let mut cmd_line = String::new();
+        for (key, value) in env {
+            cmd_line.push_str(&format!(
+                "export {}='{}'; ",
+                key,
+                value.replace("'", "'\\''")
+            ));
+        }
+        if let Some(dir) = cwd {
+            cmd_line.push_str(&format!("cd {} && ", dir));
+        }
+        cmd_line.push_str(program);
+        for arg in args {
+            cmd_line.push(' ');
+            cmd_line.push_str(arg); // TODO: escape
+        }
+
+        debug!("Executing SSH command (captured): {}", cmd_line);
+
+        // Keep the PTY: it keeps tool output line-buffered (better for live
+        // display) and merges stdout/stderr into a single ordered stream,
+        // which ssh2 cannot read concurrently anyway.
+        channel
+            .request_pty("xterm", None, None)
+            .map_err(|e| io::Error::other(format!("Failed to request PTY: {}", e)))?;
+
+        channel.exec(&cmd_line).map_err(io::Error::other)?;
+
+        let mut stdout_stream = channel.stream(0);
+        pump(&mut stdout_stream, Stream::Stdout, sink.as_ref());
+
+        channel.wait_close().map_err(io::Error::other)?;
+
+        let code = channel.exit_status().unwrap_or(-1);
+        Ok(ExitStatus::from_raw(code))
+    }
+
     fn run_output(
         &self,
         program: &str,

src/context/unshare.rs

@@ -1,4 +1,4 @@
-use super::api::{Context, ContextCommand, ContextDriver};
+use super::api::{Context, ContextCommand, ContextDriver, LineSink};
 use log::debug;
 use std::fs;
 use std::io;
@@ -259,6 +259,22 @@ impl ContextDriver for UnshareDriver {
         self.command(program, args, env, cwd).status()
     }
 
+    fn run_captured(
+        &self,
+        program: &str,
+        args: &[String],
+        env: &[(String, String)],
+        cwd: Option<&str>,
+        sink: Arc<dyn LineSink>,
+    ) -> io::Result<std::process::ExitStatus> {
+        // Forward the sink to the wrapping command: the parent context driver
+        // is responsible for the actual capture (and may chain further, e.g.
+        // unshare over ssh).
+        let mut cmd = self.command(program, args, env, cwd);
+        cmd.capture(sink);
+        cmd.status()
+    }
+
     fn run_output(
         &self,
         program: &str,