tawasuyu/brahman

1//! `shuma-exec` — ejecución de comandos del shell con salida en streaming.
2//!
3//! Lanza una línea de comandos en un shell (`bash -c …`) dentro de un
4//! directorio, y entrega su salida **a medida que ocurre**: cada línea
5//! de stdout o stderr llega como un [`RunEvent`] por un canal, sin
6//! esperar a que el proceso termine.
7//!
8//! Esto es lo que `sandokan` no hace: el orquestador es poll-based y
9//! orquesta *Cards* de brahman (entidades aisladas y supervisadas). El
10//! shell, en cambio, corre líneas de shell ad-hoc y necesita ver la
11//! salida fluir. Dos capas distintas, a propósito.
12//!
13//! **Captura acotada.** Para no cargar en RAM un stream de gigabytes, la
14//! captura tiene un límite de bytes ([`CommandSpec::capture_limit`]):
15//! pasado el límite se emite [`RunEvent::Truncated`] una vez y el resto
16//! se **descarta** — pero el pipe se sigue drenando, así el proceso no
17//! se bloquea y termina normal.
18//!
19//! **Reproceso.** [`CommandSpec::stdin_data`] alimenta un texto por la
20//! entrada estándar del proceso: permite reprocesar la salida capturada
21//! de un comando previo con otra herramienta, sin volver a correr el
22//! comando original.
23//!
24//! El crate es agnóstico de frontend: el proceso y sus lectores corren
25//! en hilos; el consumidor (shell GPUI o TUI) drena el canal cuando
26//! quiere — sin `async`, sin acoplarse a ningún runtime.
27
28#![forbid(unsafe_code)]
29
30use std::io::{BufRead, BufReader, Read, Write};
31use std::process::{Child, Command, Stdio};
32use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
33use std::sync::mpsc::{self, Receiver, Sender, TryRecvError};
34use std::sync::{Arc, Mutex};
35use std::thread::JoinHandle;
36
37/// Qué ejecutar: una línea de comandos, en un directorio, con un shell.
38#[derive(Debug, Clone)]
39pub struct CommandSpec {
40    /// La línea completa — se pasa como `shell -c "<line>"`.
41    pub line: String,
42    /// Directorio de trabajo del proceso.
43    pub cwd: String,
44    /// Programa de shell — `"bash"`, `"sh"`, `"fish"`…
45    pub shell: String,
46    /// Tope de bytes a capturar; `0` = sin límite. Pasado el tope, la
47    /// salida se descarta (se emite [`RunEvent::Truncated`]).
48    pub capture_limit: usize,
49    /// Texto a alimentar por stdin — para reprocesar una salida previa.
50    pub stdin_data: Option<String>,
51}
52
53impl CommandSpec {
54    /// Spec con `bash` como shell, sin límite ni stdin.
55    pub fn bash(line: impl Into<String>, cwd: impl Into<String>) -> Self {
56        Self {
57            line: line.into(),
58            cwd: cwd.into(),
59            shell: "bash".into(),
60            capture_limit: 0,
61            stdin_data: None,
62        }
63    }
64
65    /// Fija el tope de captura en bytes (encadenable).
66    pub fn with_limit(mut self, bytes: usize) -> Self {
67        self.capture_limit = bytes;
68        self
69    }
70
71    /// Alimenta `data` por la entrada estándar del proceso (encadenable).
72    pub fn with_stdin(mut self, data: impl Into<String>) -> Self {
73        self.stdin_data = Some(data.into());
74        self
75    }
76}
77
78/// Un evento de la ejecución de un comando.
79#[derive(Debug, Clone, PartialEq, Eq)]
80pub enum RunEvent {
81    /// Una línea de salida estándar.
82    Stdout(String),
83    /// Una línea de salida de error.
84    Stderr(String),
85    /// La captura alcanzó su tope; lo que sigue se descarta.
86    Truncated,
87    /// El proceso terminó con este código de salida.
88    Exited(i32),
89    /// El proceso no pudo siquiera lanzarse.
90    Failed(String),
91}
92
93impl RunEvent {
94    /// `true` si el evento cierra la ejecución (`Exited` o `Failed`).
95    pub fn is_terminal(&self) -> bool {
96        matches!(self, RunEvent::Exited(_) | RunEvent::Failed(_))
97    }
98}
99
100/// Asa de un comando en ejecución. El consumidor la conserva y drena sus
101/// eventos cuando le conviene.
102pub struct RunHandle {
103    rx: Receiver<RunEvent>,
104    finished: bool,
105    /// El proceso, compartido con su hilo coordinador para poder matarlo.
106    child: Arc<Mutex<Option<Child>>>,
107}
108
109impl RunHandle {
110    /// Mata el proceso (envía la señal de terminación). No hace nada si
111    /// el proceso ya terminó o nunca llegó a lanzarse.
112    pub fn kill(&self) {
113        if let Ok(mut guard) = self.child.lock() {
114            if let Some(c) = guard.as_mut() {
115                let _ = c.kill();
116            }
117        }
118    }
119
120    /// Drena todos los eventos disponibles ahora mismo, sin bloquear.
121    /// Marca el asa como terminada al ver un evento terminal.
122    pub fn try_events(&mut self) -> Vec<RunEvent> {
123        let mut out = Vec::new();
124        loop {
125            match self.rx.try_recv() {
126                Ok(ev) => {
127                    if ev.is_terminal() {
128                        self.finished = true;
129                    }
130                    out.push(ev);
131                }
132                Err(TryRecvError::Empty) => break,
133                Err(TryRecvError::Disconnected) => {
134                    self.finished = true;
135                    break;
136                }
137            }
138        }
139        out
140    }
141
142    /// Bloquea hasta que el proceso termine y devuelve todos sus eventos
143    /// en orden. Pensado para tests y para usos sincrónicos.
144    pub fn wait_all(&mut self) -> Vec<RunEvent> {
145        let mut out = Vec::new();
146        while let Ok(ev) = self.rx.recv() {
147            let terminal = ev.is_terminal();
148            out.push(ev);
149            if terminal {
150                self.finished = true;
151            }
152        }
153        self.finished = true;
154        out
155    }
156
157    /// `true` si ya se observó el evento terminal.
158    pub fn is_finished(&self) -> bool {
159        self.finished
160    }
161}
162
163/// Lanza un hilo lector de un flujo. Cuenta los bytes contra `counter`;
164/// pasado `limit` emite `Truncated` una vez (vía `announced`) y descarta
165/// el resto, pero **sigue drenando** el pipe para no bloquear al proceso.
166fn spawn_reader<R: Read + Send + 'static>(
167    stream: R,
168    tx: Sender<RunEvent>,
169    make: fn(String) -> RunEvent,
170    limit: usize,
171    counter: Arc<AtomicUsize>,
172    announced: Arc<AtomicBool>,
173) -> JoinHandle<()> {
174    std::thread::spawn(move || {
175        for line in BufReader::new(stream).lines().map_while(Result::ok) {
176            let total = counter.fetch_add(line.len() + 1, Ordering::Relaxed) + line.len() + 1;
177            if limit != 0 && total > limit {
178                if !announced.swap(true, Ordering::Relaxed) {
179                    let _ = tx.send(RunEvent::Truncated);
180                }
181                continue; // descarta, pero sigue leyendo el pipe
182            }
183            if tx.send(make(line)).is_err() {
184                break;
185            }
186        }
187    })
188}
189
190/// Lanza `spec` y devuelve un [`RunHandle`] desde el que drenar la
191/// salida. La función vuelve de inmediato: el proceso corre en hilos.
192pub fn run(spec: &CommandSpec) -> RunHandle {
193    let (tx, rx) = mpsc::channel();
194    let spec = spec.clone();
195    let child_cell: Arc<Mutex<Option<Child>>> = Arc::new(Mutex::new(None));
196    let cell = Arc::clone(&child_cell);
197
198    std::thread::spawn(move || {
199        let stdin_mode = if spec.stdin_data.is_some() {
200            Stdio::piped()
201        } else {
202            Stdio::null()
203        };
204        let spawned = Command::new(&spec.shell)
205            .arg("-c")
206            .arg(&spec.line)
207            .current_dir(&spec.cwd)
208            .stdin(stdin_mode)
209            .stdout(Stdio::piped())
210            .stderr(Stdio::piped())
211            .spawn();
212
213        let mut child = match spawned {
214            Ok(c) => c,
215            Err(e) => {
216                let _ = tx.send(RunEvent::Failed(e.to_string()));
217                return;
218            }
219        };
220
221        // Si hay datos para reprocesar, se escriben por stdin en su
222        // propio hilo (la escritura puede bloquear hasta que el proceso
223        // consuma); al terminar, `stdin` se cierra → EOF.
224        if let Some(data) = spec.stdin_data.clone() {
225            if let Some(mut stdin) = child.stdin.take() {
226                std::thread::spawn(move || {
227                    let _ = stdin.write_all(data.as_bytes());
228                });
229            }
230        }
231
232        let stdout = child.stdout.take();
233        let stderr = child.stderr.take();
234        // Comparte el proceso para que `RunHandle::kill` pueda alcanzarlo.
235        if let Ok(mut g) = cell.lock() {
236            *g = Some(child);
237        }
238
239        // Contador de bytes compartido: el tope vale para stdout+stderr.
240        let counter = Arc::new(AtomicUsize::new(0));
241        let announced = Arc::new(AtomicBool::new(false));
242        let limit = spec.capture_limit;
243
244        let out_reader = stdout.map(|s| {
245            spawn_reader(
246                s,
247                tx.clone(),
248                RunEvent::Stdout,
249                limit,
250                Arc::clone(&counter),
251                Arc::clone(&announced),
252            )
253        });
254        let err_reader = stderr.map(|s| {
255            spawn_reader(
256                s,
257                tx.clone(),
258                RunEvent::Stderr,
259                limit,
260                Arc::clone(&counter),
261                Arc::clone(&announced),
262            )
263        });
264
265        // Los lectores terminan cuando el proceso cierra sus pipes —sea
266        // por fin natural o por `kill`—; recién entonces se cosecha.
267        if let Some(h) = out_reader {
268            let _ = h.join();
269        }
270        if let Some(h) = err_reader {
271            let _ = h.join();
272        }
273        let code = cell
274            .lock()
275            .ok()
276            .and_then(|mut g| g.as_mut().and_then(|c| c.wait().ok()))
277            .and_then(|s| s.code())
278            .unwrap_or(-1);
279        let _ = tx.send(RunEvent::Exited(code));
280    });
281
282    RunHandle { rx, finished: false, child: child_cell }
283}
284
285#[cfg(test)]
286mod tests {
287    use super::*;
288
289    /// `sh` está en cualquier entorno POSIX — más portable que bash
290    /// para los tests.
291    fn sh(line: &str) -> CommandSpec {
292        CommandSpec { shell: "sh".into(), ..CommandSpec::bash(line, ".") }
293    }
294
295    #[test]
296    fn captures_stdout_and_exit_code() {
297        let mut h = run(&sh("echo hola-mundo"));
298        let events = h.wait_all();
299        assert!(events.contains(&RunEvent::Stdout("hola-mundo".into())));
300        assert!(events.contains(&RunEvent::Exited(0)));
301        assert!(h.is_finished());
302    }
303
304    #[test]
305    fn captures_stderr() {
306        let mut h = run(&sh("echo problema 1>&2"));
307        let events = h.wait_all();
308        assert!(events.contains(&RunEvent::Stderr("problema".into())));
309    }
310
311    #[test]
312    fn nonzero_exit_is_reported() {
313        let mut h = run(&sh("exit 3"));
314        let events = h.wait_all();
315        assert!(events.contains(&RunEvent::Exited(3)));
316    }
317
318    #[test]
319    fn multiple_output_lines_arrive_in_order() {
320        let mut h = run(&sh("echo uno; echo dos; echo tres"));
321        let lines: Vec<String> = h
322            .wait_all()
323            .into_iter()
324            .filter_map(|e| match e {
325                RunEvent::Stdout(l) => Some(l),
326                _ => None,
327            })
328            .collect();
329        assert_eq!(lines, vec!["uno", "dos", "tres"]);
330    }
331
332    #[test]
333    fn pipes_run_through_the_shell() {
334        let mut h = run(&sh("printf 'b\\na\\nc\\n' | sort"));
335        let lines: Vec<String> = h
336            .wait_all()
337            .into_iter()
338            .filter_map(|e| match e {
339                RunEvent::Stdout(l) => Some(l),
340                _ => None,
341            })
342            .collect();
343        assert_eq!(lines, vec!["a", "b", "c"]);
344    }
345
346    #[test]
347    fn missing_shell_fails_gracefully() {
348        let spec = CommandSpec { shell: "/no/existe/shell-xyz".into(), ..sh("echo x") };
349        let mut h = run(&spec);
350        let events = h.wait_all();
351        assert!(matches!(events.first(), Some(RunEvent::Failed(_))));
352    }
353
354    #[test]
355    fn terminal_event_detection() {
356        assert!(RunEvent::Exited(0).is_terminal());
357        assert!(RunEvent::Failed("x".into()).is_terminal());
358        assert!(!RunEvent::Stdout("x".into()).is_terminal());
359        assert!(!RunEvent::Truncated.is_terminal());
360    }
361
362    #[test]
363    fn kill_stops_a_long_running_process() {
364        let mut h = run(&sh("sleep 30"));
365        std::thread::sleep(std::time::Duration::from_millis(250));
366        h.kill();
367        let events = h.wait_all();
368        assert!(events.last().map(|e| e.is_terminal()).unwrap_or(false));
369        assert!(h.is_finished());
370    }
371
372    #[test]
373    fn capture_limit_truncates_but_process_finishes() {
374        // 20.000 líneas, pero la captura se corta a ~400 bytes.
375        let mut h = run(&sh("seq 1 20000").with_limit(400));
376        let events = h.wait_all();
377        // Se anunció el truncado…
378        assert!(events.contains(&RunEvent::Truncated));
379        // …pero el proceso terminó normal (no se bloqueó).
380        assert!(events.contains(&RunEvent::Exited(0)));
381        // Y lo capturado quedó acotado.
382        let captured = events
383            .iter()
384            .filter(|e| matches!(e, RunEvent::Stdout(_)))
385            .count();
386        assert!(captured < 20000, "la salida quedó acotada");
387    }
388
389    #[test]
390    fn no_limit_captures_everything() {
391        let mut h = run(&sh("seq 1 500")); // capture_limit = 0
392        let events = h.wait_all();
393        assert!(!events.contains(&RunEvent::Truncated));
394        let n = events.iter().filter(|e| matches!(e, RunEvent::Stdout(_))).count();
395        assert_eq!(n, 500);
396    }
397
398    #[test]
399    fn stdin_data_is_fed_to_the_process() {
400        // `cat` devuelve por stdout lo que recibe por stdin — es el
401        // reproceso más simple: tomar una salida y pasarla a otro filtro.
402        let mut h = run(&sh("cat").with_stdin("alfa\nbeta\ngamma"));
403        let lines: Vec<String> = h
404            .wait_all()
405            .into_iter()
406            .filter_map(|e| match e {
407                RunEvent::Stdout(l) => Some(l),
408                _ => None,
409            })
410            .collect();
411        assert_eq!(lines, vec!["alfa", "beta", "gamma"]);
412    }
413
414    #[test]
415    fn stdin_data_reprocessed_by_a_filter() {
416        let mut h = run(&sh("grep beta").with_stdin("alfa\nbeta\nbetabel\ngamma"));
417        let lines: Vec<String> = h
418            .wait_all()
419            .into_iter()
420            .filter_map(|e| match e {
421                RunEvent::Stdout(l) => Some(l),
422                _ => None,
423            })
424            .collect();
425        assert_eq!(lines, vec!["beta", "betabel"]);
426    }
427}