From 095bbd33098642ab9fb1925b5916288b951561b4 Mon Sep 17 00:00:00 2001 From: Marut K Date: Fri, 13 Nov 2026 22:22:26 +0700 Subject: [PATCH] feat: report task and child-program OSC 7501 status --- src/apps/cli/app.rs | 22 ++- src/apps/exec.rs | 8 +- src/apps/mod.rs | 1 + src/apps/program_status.rs | 311 ++++++++++++++++++++++++++++++++++++ src/apps/task.rs | 64 +++++++- src/apps/tui/markdown.rs | 6 +- src/apps/tui/preview/mod.rs | 2 +- src/apps/tui/tasks.rs | 11 +- 8 files changed, 392 insertions(+), 33 deletions(-) create mode 100644 src/apps/program_status.rs diff --git a/src/apps/cli/app.rs b/src/apps/cli/app.rs index 2ac45d8..0d18f1b 100644 --- a/src/apps/cli/app.rs +++ b/src/apps/cli/app.rs @@ -261,7 +261,7 @@ impl App { let mut outputs = self.outputs.borrow_mut(); let state = outputs .entry(id) - .or_insert_with(|| task::Task::new(pty_cols, pty_rows, 1024)); + .or_insert_with(|| task::Task::new(id, pty_cols, pty_rows, 1024)); let command = exec::run_code( code, size, @@ -395,19 +395,15 @@ impl App { }; let was_alt = state.is_alternate_screen(); - exec::handle_stream(state, stream); - let now_alt = state.is_alternate_screen(); - - // When a full-screen TUI app (vim, less, htop) is active in the - // alternate screen buffer, write its PTY output directly to the - // real terminal so escape sequences (cursor positioning, alt - // screen entry/exit) are processed natively. - if let Stream::Out(s) = stream { - if self.batch_output.is_terminal() && focused && (was_alt || now_alt) { - let _ = io::stdout().write_all(s.as_bytes()); - let _ = io::stdout().flush(); + match stream { + Stream::Out(text) => { + state.process_output(text, self.batch_output.is_terminal() && focused); + } + _ => { + exec::handle_stream(state, stream); } } + let now_alt = state.is_alternate_screen(); // Resize PTY when entering/exiting alternate screen so TUI apps // get the full terminal dimensions and line-oriented programs get @@ -1170,7 +1166,7 @@ echo world } fn insert_completed_output(app: &App, id: CodeId, text: &str, exit_code: i32) { - let mut output = task::Task::new(80, 24, 1024); + let mut output = task::Task::new(id, 80, 24, 1024); exec::handle_stream(&mut output, &Stream::Out(text.to_string())); exec::handle_stream(&mut output, &Stream::Exit(exit_code)); exec::handle_stream(&mut output, &Stream::End); diff --git a/src/apps/exec.rs b/src/apps/exec.rs index 37a8b5a..4fefd25 100644 --- a/src/apps/exec.rs +++ b/src/apps/exec.rs @@ -36,6 +36,7 @@ pub fn run_code( .parser .parse(&format!("Failed to determine current directory: {e}")); output.done = true; + output.report_status(); return None; } }, @@ -44,11 +45,13 @@ pub fn run_code( Ok(exec) => { let rx = exec.receiver(); output.execution = Some(exec); + output.report_status(); Some(rx) } Err(err) => { output.parser.parse(&err.to_string()); output.done = true; + output.report_status(); None } } @@ -60,9 +63,7 @@ pub fn run_code( pub fn handle_stream(output: &mut Task, stream: &Stream) -> bool { let mut force_rebuild = false; match stream { - Stream::Out(s) => { - output.parser.parse(s); - } + Stream::Out(s) => output.process_output(s, false), Stream::Env(envs) => { output.captured_envs = Some(envs.clone()); } @@ -77,6 +78,7 @@ pub fn handle_stream(output: &mut Task, stream: &Stream) -> bool { Stream::End => { force_rebuild = !output.done; output.done = true; + output.report_status(); } } force_rebuild diff --git a/src/apps/mod.rs b/src/apps/mod.rs index fb938ab..9bdbe10 100644 --- a/src/apps/mod.rs +++ b/src/apps/mod.rs @@ -6,6 +6,7 @@ pub mod config; pub mod exec; pub mod navigation; pub mod picker; +mod program_status; pub mod task; mod theme; pub mod tui; diff --git a/src/apps/program_status.rs b/src/apps/program_status.rs new file mode 100644 index 0000000..9f7ba26 --- /dev/null +++ b/src/apps/program_status.rs @@ -0,0 +1,311 @@ +//! Child-process OSC 7501 status framing and ID scoping. +//! +//! Frames OSC 7501 sequences across PTY chunks, scopes child IDs under the +//! owning block, and replies to capability queries. Status sequences are +//! consumed while all other bytes pass through unchanged. + +use std::{ + fmt::{Arguments, Write as _}, + io::{self, IsTerminal, Write}, +}; + +use upmd_parser::CodeId; + +const MAX_SEQUENCE: usize = 4096; +const MAX_ID: usize = 128; +const MAX_ID_SEGMENTS: usize = 5; +const OSC_PREFIX: &[u8] = b"\x1b]7501;"; +const MAX_BODY: usize = MAX_SEQUENCE - OSC_PREFIX.len() - 1; +const QUERY_REPLY: &[u8] = + b"\x1b]7501;?:states=idle,working,done,blocked,error:kinds=permission,question,auth\x1b\\"; + +/// Per-execution child status framing, forwarding, and record cleanup. +#[derive(Default)] +pub(crate) struct ChildStatus { + state: ParseState, + input: Vec, + reported_id: Option, +} + +enum ChildEvent<'a> { + Output(&'a [u8]), + Query, + /// An OSC sequence with a block-scoped ID; other fields are passed through. + Report(String), +} + +#[derive(Clone, Copy, Default)] +enum ParseState { + #[default] + Ground, + /// Matched `&OSC_PREFIX[..len]` (`1..OSC_PREFIX.len()`). + Prefix(usize), + /// Inside an `OSC 7501` body. + Status { escaped: bool, discard: bool }, +} + +/// Returns whether a terminal write was attempted. Reporting is best-effort. +pub(crate) fn write(text: Arguments<'_>) -> bool { + let stdout = io::stdout(); + if !stdout.is_terminal() { + return false; + } + let mut stdout = stdout.lock(); + let _ = stdout.write_fmt(text); + let _ = stdout.flush(); + true +} + +impl ChildStatus { + /// Consumes OSC 7501 sequences from a PTY chunk and optionally forwards + /// the remaining bytes to `stdout`. + /// + /// While `active`, valid status reports are scoped under `upmd/block-{id}/child` and + /// written to the outer terminal, and capability queries (`?`) invoke `reply`. + pub(crate) fn feed( + &mut self, + id: CodeId, + bytes: &[u8], + active: bool, + forward_to_stdout: bool, + mut reply: impl FnMut(&[u8]), + ) { + let mut stdout = io::stdout(); + let mut reported = false; + self.parse(id, bytes, active, |event| match event { + ChildEvent::Output(bytes) if forward_to_stdout => { + let _ = stdout.write_all(bytes); + } + ChildEvent::Output(_) => {} + ChildEvent::Query => reply(QUERY_REPLY), + ChildEvent::Report(sequence) => { + reported |= write(format_args!("{sequence}")); + } + }); + if forward_to_stdout { + let _ = stdout.flush(); + } + if reported { + self.reported_id = Some(id); + } + } + + pub(crate) fn reset(&mut self) { + self.clear(); + self.state = ParseState::Ground; + self.input.clear(); + } + + /// Clears records without losing framing for output queued after process exit. + pub(crate) fn clear(&mut self) { + if let Some(id) = self.reported_id.take() { + write(format_args!( + "\x1b]7501;state=clear:id=upmd/block-{id}/child\x1b\\" + )); + } + } + + fn parse( + &mut self, + id: CodeId, + bytes: &[u8], + active: bool, + mut emit: impl FnMut(ChildEvent<'_>), + ) { + let mut raw_start = 0; + let mut index = 0; + while index < bytes.len() { + let byte = bytes[index]; + match self.state { + ParseState::Ground => { + let Some(offset) = bytes[index..].iter().position(|&b| b == 0x1b) else { + break; + }; + let esc = index + offset; + if raw_start < esc { + emit(ChildEvent::Output(&bytes[raw_start..esc])); + } + raw_start = esc + 1; + index = esc + 1; + self.state = ParseState::Prefix(1); + continue; + } + // Withhold bytes while matching `\x1b]7501;`. + ParseState::Prefix(len) => { + raw_start = index + 1; + if byte == OSC_PREFIX[len] { + if len + 1 == OSC_PREFIX.len() { + self.input.clear(); + self.state = ParseState::Status { + escaped: false, + discard: !active, + }; + } else { + self.state = ParseState::Prefix(len + 1); + } + } else { + // Mismatch: flush the withheld prefix and re-check this byte in Ground. + emit(ChildEvent::Output(&OSC_PREFIX[..len])); + raw_start = index; + self.state = ParseState::Ground; + continue; + } + } + // Consume OSC 7501 body until BEL (`\x07`) or ST (`\x1b\`). + ParseState::Status { + escaped, + mut discard, + } => { + raw_start = index + 1; + let ended = (escaped && byte == b'\\') || byte == 0x07; + // Any ESC not followed by `\` is malformed. + discard |= !active || (escaped && byte != b'\\'); + + if ended { + self.state = ParseState::Ground; + if !discard { + if self.input.trim_ascii() == b"?" { + emit(ChildEvent::Query); + } else if let Some(report) = scope_id(id, &self.input) { + emit(ChildEvent::Report(report)); + } + } + } else { + // Buffer payload bytes (hold `0x1b` in case it starts `\x1b\`). + if !discard && byte != 0x1b { + if self.input.len() < MAX_BODY { + self.input.push(byte); + } else { + discard = true; + } + } + self.state = ParseState::Status { + escaped: byte == 0x1b, + discard, + }; + } + } + } + index += 1; + } + if raw_start < bytes.len() { + emit(ChildEvent::Output(&bytes[raw_start..])); + } + } +} + +/// Replace only ID values, preserving field order and all other body bytes. +fn scope_id(block: CodeId, input: &[u8]) -> Option { + let body = std::str::from_utf8(input).ok()?; + // Keep the forwarded body inside its OSC frame. + if body.chars().any(char::is_control) { + return None; + } + let mut output = String::from("\x1b]7501;"); + let mut has_id = false; + for (i, pair) in body.split(':').enumerate() { + if i > 0 { + output.push(':'); + } + match pair.split_once('=') { + Some((key, value)) if key.trim() == "id" => { + let id = value.trim(); + let scoped = format!("upmd/block-{block}/child/{id}"); + if !valid_id(id) || scoped.len() > MAX_ID { + return None; + } + let _ = write!(output, "{key}={scoped}"); + has_id = true; + } + _ => output.push_str(pair), + } + } + if !has_id { + let separator = if body.is_empty() { "" } else { ":" }; + let _ = write!(output, "{separator}id=upmd/block-{block}/child"); + } + output.push_str("\x1b\\"); + (output.len() <= MAX_SEQUENCE).then_some(output) +} + +fn valid_id(id: &str) -> bool { + id.split('/').count() <= MAX_ID_SEGMENTS + && id.split('/').all(|s| { + (1..=32).contains(&s.len()) + && s.bytes() + .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'.' | b'+' | b'-')) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn collect(decoder: &mut ChildStatus, bytes: &[u8]) -> (Vec, Vec) { + let mut output = Vec::new(); + let mut events = Vec::new(); + decoder.parse(1, bytes, true, |event| match event { + ChildEvent::Output(bytes) => output.extend_from_slice(bytes), + ChildEvent::Query => events.push("query".to_owned()), + ChildEvent::Report(report) => events.push(report), + }); + (output, events) + } + + #[test] + fn scopes_ids_and_rejects_invalid_reports() { + let cases = [ + ("state=clear", Some("state=clear:id=upmd/block-1/child")), + ( + "id=old:state=done: id = worker/sub ", + Some("id=upmd/block-1/child/old:state=done: id =upmd/block-1/child/worker/sub"), + ), + ( + "state=done:future=literal id=untouched:id=worker:msg=SGk=::", + Some("state=done:future=literal id=untouched:id=upmd/block-1/child/worker:msg=SGk=::"), + ), + ("state=clear:id=", None), + ("state=clear:id=bad%2Fid", None), + ("state=clear:id=a//b", None), + ("state=clear:id=a/b/c/d/e/f", None), + ("state=clear\x18outside", None), + ]; + for (body, expected) in cases { + let expected = expected.map(|b| format!("\x1b]7501;{b}\x1b\\")); + assert_eq!(scope_id(1, body.as_bytes()), expected, "{body}"); + } + } + + #[test] + fn frames_status_and_filters_passthrough_across_chunks() { + let input = b"before\x1b]7501;?\x07\x1b]0;title\x07\x1b]75\x07\x1b\x1b[0m\x1b]7501;state=blocked:id=worker\x1b\\after"; + let expected_out = b"before\x1b]0;title\x07\x1b]75\x07\x1b\x1b[0mafter"; + let expected_events = [ + "query", + "\x1b]7501;state=blocked:id=upmd/block-1/child/worker\x1b\\", + ]; + + for split in 0..=input.len() { + let mut decoder = ChildStatus::default(); + let (mut out, mut events) = collect(&mut decoder, &input[..split]); + let (out2, events2) = collect(&mut decoder, &input[split..]); + out.extend(out2); + events.extend(events2); + assert_eq!(out, expected_out, "split {split}"); + assert_eq!(events, expected_events, "split {split}"); + } + + // Oversized sequences and inactive (post-exit) streams still strip OSC 7501 from output. + let oversized = format!( + "\x1b]7501;msg={}\x07\x1b]7501;?\x07tail", + "a".repeat(MAX_SEQUENCE) + ); + let mut decoder = ChildStatus::default(); + let mut out = Vec::new(); + decoder.parse(1, oversized.as_bytes(), false, |event| match event { + ChildEvent::Output(b) => out.extend_from_slice(b), + _ => panic!("inactive event"), + }); + assert_eq!(out, b"tail"); + } +} diff --git a/src/apps/task.rs b/src/apps/task.rs index f7c89c1..bd20eb2 100644 --- a/src/apps/task.rs +++ b/src/apps/task.rs @@ -2,11 +2,13 @@ //! //! [`Task`] combines the low-level process handle and PTY parser with the //! viewport state (scroll positions, dirty flag) that the TUI and CLI need -//! for rendering. A `Task` is created when a code block is executed and -//! persists until the document is reloaded. +//! for rendering. It also reports its process lifecycle to the outer terminal. +//! A `Task` is created when a code block is executed and persists until the +//! document is reloaded. -use crate::apps::config::Envs; +use crate::apps::{config::Envs, program_status}; use crate::runner; +use upmd_parser::CodeId; use upmd_pty::parser::Parser as PtyParser; /// Lifecycle status of a code block execution. @@ -23,7 +25,9 @@ pub enum TaskStatus { /// Full state of one code-block execution: PTY output, scroll, and flags. pub struct Task { + pub id: CodeId, pub parser: PtyParser, + child_status: program_status::ChildStatus, pub execution: Option, pub done: bool, pub exit_code: Option, @@ -38,9 +42,11 @@ pub struct Task { } impl Task { - pub fn new(cols: u16, rows: u16, scrollback: usize) -> Self { + pub fn new(id: CodeId, cols: u16, rows: u16, scrollback: usize) -> Self { Self { + id, parser: PtyParser::new(rows, cols, scrollback), + child_status: Default::default(), execution: None, done: false, exit_code: None, @@ -74,8 +80,49 @@ impl Task { } } + /// Reports this task's current process status to the outer terminal. + pub(crate) fn report_status(&mut self) { + if self.done { + self.child_status.clear(); + } + self.emit_status(self.status()); + } + + fn emit_status(&self, status: TaskStatus) { + let state = match status { + TaskStatus::Idle => "idle", + TaskStatus::Running => "working", + TaskStatus::Success => "done", + TaskStatus::Error => "error", + }; + program_status::write(format_args!( + "\x1b]7501;state={state}:app=upmd:id=upmd/block-{}\x1b\\", + self.id + )); + } + + /// Parses PTY output, relays child OSC 7501 status/queries, and optionally + /// forwards filtered alternate-screen output to the host terminal (CLI mode). + pub(crate) fn process_output(&mut self, text: &str, forward_alt_screen: bool) { + let was_alt = self.is_alternate_screen(); + self.parser.parse(text); + let forward_to_stdout = forward_alt_screen && (was_alt || self.is_alternate_screen()); + self.child_status.feed( + self.id, + text.as_bytes(), + !self.done, + forward_to_stdout, + |reply| { + if let Some(execution) = self.execution.as_mut() { + let _ = execution.process_mut().write(reply); + } + }, + ); + } + /// Resets execution state while preserving PTY parser dimensions. pub fn reset(&mut self) { + self.child_status.reset(); self.execution = None; self.done = false; self.exit_code = None; @@ -124,3 +171,12 @@ impl Task { self.parser.scroll(n); } } + +impl Drop for Task { + fn drop(&mut self) { + self.child_status.clear(); + if self.running() { + self.emit_status(TaskStatus::Idle); + } + } +} diff --git a/src/apps/tui/markdown.rs b/src/apps/tui/markdown.rs index 7c81778..122c4d9 100644 --- a/src/apps/tui/markdown.rs +++ b/src/apps/tui/markdown.rs @@ -1832,7 +1832,7 @@ mod tests { #[test] fn test_failed_start_code_info_renders_error_status_and_gutter() { let mut outputs = HashMap::new(); - let mut output = Task::new(80, 24, 500); + let mut output = Task::new(1, 80, 24, 500); output.done = true; output.exit_code = None; outputs.insert(1, output); @@ -1873,7 +1873,7 @@ mod tests { /// Creates a [`Task`] pre-loaded with 50 lines of output for use in /// inline-scroll snapshot tests. fn output_task() -> Task { - let mut buf = Task::new(40, 24, 500); + let mut buf = Task::new(1, 40, 24, 500); for i in 0..50 { buf.parser .parse(&format!("This is output line number {i}\n")); @@ -1957,7 +1957,7 @@ mod tests { #[test] fn test_inline_scroll_short_output() { // Output shorter than inline_max_lines. All lines visible at any scroll. - let mut buf = Task::new(40, 24, 500); + let mut buf = Task::new(1, 40, 24, 500); for i in 0..5 { buf.parser.parse(&format!("short line {i}\n")); } diff --git a/src/apps/tui/preview/mod.rs b/src/apps/tui/preview/mod.rs index 6a4fe41..5f640b0 100644 --- a/src/apps/tui/preview/mod.rs +++ b/src/apps/tui/preview/mod.rs @@ -1822,7 +1822,7 @@ mod tests { let doc = upmd_parser::new().parse("```bash\necho hello\n```"); let theme = Theme::new("base16-ocean.dark", false); let mut outputs = HashMap::new(); - let mut output = Task::new(80, 24, 500); + let mut output = Task::new(1, 80, 24, 500); output.done = true; output.exit_code = Some(0); outputs.insert(1, output); diff --git a/src/apps/tui/tasks.rs b/src/apps/tui/tasks.rs index 5903503..271c3d9 100644 --- a/src/apps/tui/tasks.rs +++ b/src/apps/tui/tasks.rs @@ -38,16 +38,9 @@ impl Tasks { let task = self .outputs .entry(code.id) - .or_insert_with(|| Task::new(size.width, size.height, 500)); - task.reset(); + .or_insert_with(|| Task::new(code.id, size.width, size.height, 500)); - match exec::run_code(code, size, envs, capture_state, binaries, task, working_dir) { - Some(rx) => { - task.dirty = true; - Some(rx) - } - None => None, - } + exec::run_code(code, size, envs, capture_state, binaries, task, working_dir) } /// Handles a stream message from a running task.