From 6f17ae0f8770b0e2b9391fba66c3968d59489d7e Mon Sep 17 00:00:00 2001 From: daniel Date: Mon, 21 Sep 2026 07:04:30 -0700 Subject: [PATCH 1/3] perf(tui): reduce streaming and transcript rendering latency --- src/tui/app.rs | 191 +++++++++++-- src/tui/markdown.rs | 663 ++++++++++++++++++++++++++++++++++++------- src/tui/mod.rs | 86 +++--- src/tui/scheduler.rs | 446 +++++++++++++++++++++++++++++ src/tui/ui.rs | 517 +++++++++++++++++++++++++++++++-- 5 files changed, 1720 insertions(+), 183 deletions(-) create mode 100644 src/tui/scheduler.rs diff --git a/src/tui/app.rs b/src/tui/app.rs index e352850..a22e542 100644 --- a/src/tui/app.rs +++ b/src/tui/app.rs @@ -669,6 +669,9 @@ pub(super) struct CachedTranscriptBlock { pub revision: u64, pub rows: Vec, pub images: Vec, + pub stable_prefix: String, + pub prefix_rows: usize, + pub prefix_lines: usize, } /// What the client is doing right now. @@ -1277,6 +1280,13 @@ impl App { } fn mark_block_dirty(&mut self, index: usize) { + // Hit ranges describe the last presented frame, not the newly mutated + // source. Only inspect the bounded viewport; preserve selection state. + for hit in &mut self.row_code { + if hit.as_ref().is_some_and(|hit| hit.block == index) { + *hit = None; + } + } if let Some(&owner) = self.tool_owners.get(&index) { self.mark_block_dirty(owner); } @@ -1665,13 +1675,17 @@ impl App { _ => Vec::new(), }) .unwrap_or_default(); - let previous_text = previous - .iter() - .filter_map(|part| match part { - AgentPart::Text(text) => Some(text.as_str()), - AgentPart::Image(_) => None, - }) - .collect::(); + let previous_text = if append { + String::new() + } else { + previous + .iter() + .filter_map(|part| match part { + AgentPart::Text(text) => Some(text.as_str()), + AgentPart::Image(_) => None, + }) + .collect::() + }; if !append { let released = previous .iter() @@ -1842,7 +1856,12 @@ impl App { match (&mut self.blocks[index], role) { (Block::User(existing), MessageRole::User) => { if append { - let last_line = existing.text.bytes().filter(|&byte| byte == b'\n').count(); + let last_line = if existing.images.is_empty() && images.is_empty() { + 0 + } else { + existing.text.bytes().filter(|&byte| byte == b'\n').count() + }; + let mut line_offset = last_line; let follows_image = existing.images.iter().any(|image| image.line == last_line); let starts_image = !images.is_empty(); @@ -1853,9 +1872,8 @@ impl App { && (follows_image || starts_image) { existing.text.push('\n'); + line_offset += 1; } - let line_offset = - existing.text.bytes().filter(|&byte| byte == b'\n').count(); let byte_offset = existing.text.len(); existing.text.push_str(&text); for image in &mut images { @@ -1974,17 +1992,17 @@ impl App { let turn_millis = self.stop_turn_timer(); self.phase = Phase::Idle; self.compacting = false; - let inherited_background: HashSet<_> = self - .tool_indices - .values() - .filter_map(|&child| self.has_background_ancestor(child).then_some(child)) - .collect(); + // Only running calls can need terminalization. Historical calls cannot + // change here, and need not walk their ancestor chains again. + let candidates: Vec<_> = self.transcript_dynamic.iter().copied().collect(); let mut finished = Vec::new(); - for (index, block) in self.blocks.iter_mut().enumerate() { - if let Block::Tool(call) = block + for index in candidates { + if self.has_background_ancestor(index) { + continue; + } + if let Block::Tool(call) = &mut self.blocks[index] && call.running() && !call.backgrounded - && !inherited_background.contains(&index) { call.status = if successful { ToolCallStatus::Completed @@ -2016,9 +2034,11 @@ impl App { /// Top-level calls that detached from their turn and are still running. pub fn background_calls(&self) -> Vec<&ToolCall> { - self.blocks + // The ordered dynamic index includes every running call, including + // detached calls; preserve transcript order without scanning history. + self.transcript_dynamic .iter() - .filter_map(|block| match block { + .filter_map(|&index| match &self.blocks[index] { Block::Tool(call) if call.parent_id.is_none() && call.backgrounded && call.running() => { @@ -8020,6 +8040,83 @@ mod tests { assert_eq!(app.latest_agent_text().as_deref(), Some("beforeafter more")); } + #[test] + fn user_appends_preserve_image_lines_and_label_byte_ranges() { + let mut app = app(); + for (text, image_line) in [ + ("α\nbeta", None), + (" plus", None), + ("[Image #1]", Some(0)), + ("after", None), + ("\n[Image #2]", Some(1)), + ] { + app.apply(Update::UserMessage { + id: "user-stream".into(), + text: text.into(), + append: true, + images: image_line + .map(|line| UserImage::new("AQID".into(), "image/png".into(), line).unwrap()) + .into_iter() + .collect(), + }); + } + let Block::User(message) = &app.blocks[0] else { + panic!("user"); + }; + assert_eq!(message.text, "α\nbeta plus\n[Image #1]\nafter\n[Image #2]"); + assert_eq!( + message + .images + .iter() + .map(|image| image.line) + .collect::>(), + [2, 4] + ); + for (image, label) in message.images.iter().zip(["[Image #1]", "[Image #2]"]) { + assert_eq!(&message.text[image.open_label.clone().unwrap()], label); + } + } + + #[test] + fn multipart_append_then_replacement_preserves_source_and_releases_images() { + let mut app = app(); + let image = UserImage::new("AQID".into(), "image/png".into(), 0).unwrap(); + app.apply(Update::AgentParts { + id: "parts".into(), + parts: vec![ + AgentPart::Text("before".into()), + AgentPart::Image(image), + AgentPart::Text("after".into()), + ], + }); + app.apply(Update::AgentMessage { + id: "parts".into(), + text: " appended".into(), + append: true, + }); + assert_eq!( + app.latest_agent_text().as_deref(), + Some("beforeafter appended") + ); + assert_eq!(app.retained_image_source_bytes, 4); + let Block::AgentParts(parts) = &app.blocks[0] else { + panic!("parts"); + }; + assert!(matches!(parts.last(), Some(AgentPart::Text(text)) if text == "after appended")); + app.apply(Update::AgentParts { + id: "parts".into(), + parts: vec![AgentPart::Text("replacement".into())], + }); + assert_eq!(app.latest_agent_text().as_deref(), Some("replacement")); + assert_eq!(app.retained_image_source_bytes, 0); + app.apply(Update::AgentMessage { + id: "parts".into(), + text: " tail".into(), + append: true, + }); + assert_eq!(app.latest_agent_text().as_deref(), Some("replacement tail")); + } + #[test] fn assistant_images_complete_upsert_replaces_stream_and_shrinks_in_place() { let mut app = app(); @@ -10031,6 +10128,60 @@ mod tests { assert!(app.transcript_dirty.contains(&0)); } + #[test] + fn finish_turn_only_terminalizes_active_foreground_calls() { + for successful in [false, true] { + let mut app = app(); + app.phase = Phase::Working; + start(&mut app, "historical", false, false); + patch(&mut app, "historical", Some(ToolCallStatus::Failed), "old"); + start(&mut app, "foreground", true, false); + start(&mut app, "foreground-child", false, false); + parent(&mut app, "foreground-child", "foreground"); + start(&mut app, "detached-completed", true, true); + start(&mut app, "middle", true, false); + parent(&mut app, "middle", "detached-completed"); + start(&mut app, "leaf", false, false); + parent(&mut app, "leaf", "middle"); + patch( + &mut app, + "detached-completed", + Some(ToolCallStatus::Completed), + "done", + ); + start(&mut app, "background-a", true, true); + start(&mut app, "background-b", false, true); + app.finish_turn_with_outcome(successful, None); + let expected = if successful { + ToolCallStatus::Completed + } else { + ToolCallStatus::Failed + }; + for id in ["foreground", "foreground-child"] { + assert_eq!(call(&app, id).status, expected); + assert!(call(&app, id).finished.is_some()); + } + assert_eq!(call(&app, "historical").status, ToolCallStatus::Failed); + assert_eq!( + call(&app, "detached-completed").status, + ToolCallStatus::Completed + ); + for id in ["middle", "leaf", "background-a", "background-b"] { + assert!(call(&app, id).running(), "{id}"); + } + assert_eq!( + app.background_calls() + .iter() + .map(|call| call.id.as_str()) + .collect::>(), + ["background-a", "background-b"] + ); + // Repeated idle terminal events cannot change inherited work. + app.finish_turn_with_outcome(!successful, None); + assert!(call(&app, "leaf").running()); + } + } + #[test] fn canonical_reparenting_updates_only_the_related_tree() { let mut app = app(); diff --git a/src/tui/markdown.rs b/src/tui/markdown.rs index 1fd957b..f6670e6 100644 --- a/src/tui/markdown.rs +++ b/src/tui/markdown.rs @@ -513,15 +513,129 @@ struct Link<'a> { url_range: Range, } +// Balanced-delimiter metadata is shared by every suffix searched during one +// inline parse. Incomplete streaming syntax is as important to cache as a +// successful match: a missing entry must not trigger another suffix scan. +struct DelimiterMatches { + source_len: usize, + image_labels: std::collections::HashMap, + raw_label_ends: std::collections::HashMap, + image_parentheses: std::collections::HashMap, + link_destinations: std::collections::HashMap, +} + +impl DelimiterMatches { + fn new(source: &str) -> Self { + let mut matches = Self { + source_len: source.len(), + image_labels: std::collections::HashMap::new(), + raw_label_ends: std::collections::HashMap::new(), + image_parentheses: std::collections::HashMap::new(), + link_destinations: std::collections::HashMap::new(), + }; + let mut raw_brackets = Vec::new(); + let mut image_parentheses = Vec::new(); + let mut brackets = Vec::new(); + let mut parentheses = Vec::new(); + let mut escaped = false; + for (index, byte) in source.bytes().enumerate() { + match byte { + b'[' => raw_brackets.push(index), + b']' => { + for open in raw_brackets.drain(..) { + matches.raw_label_ends.insert(open, index); + } + } + _ => {} + } + // Markdown link destinations historically count even escaped + // parentheses, whereas image alt text honors punctuation escapes. + match byte { + b'(' => parentheses.push(index), + b')' => { + if let Some(open) = parentheses.pop() { + matches.link_destinations.insert(open, index); + } + } + _ => {} + } + if escaped { + escaped = false; + if byte.is_ascii_punctuation() { + continue; + } + } + match byte { + b'\\' => escaped = true, + b'\r' | b'\n' => image_parentheses.clear(), + b'(' => image_parentheses.push(index), + b')' => { + if let Some(open) = image_parentheses.pop() { + matches.image_parentheses.insert(open, index); + } + } + b'[' => brackets.push(index), + b']' => { + if let Some(open) = brackets.pop() { + matches.image_labels.insert(open, index); + } + } + _ => {} + } + } + matches + } + + // Callers pass suffixes ending at the same source boundary. Document image + // scanning creates fresh metadata for each fence-delimited prose segment. + fn closing(&self, source: &str, opening: usize, image: bool) -> Option { + let base = self.source_len - source.len(); + let matches = if image { + &self.image_labels + } else { + &self.link_destinations + }; + matches.get(&(base + opening)).map(|end| *end - base) + } +} + /// Returns byte ranges for complete inline images and their destinations, in source order. /// Reference-style images are left literal. Destinations may contain local-path spaces; /// Markdown punctuation escapes are decoded, but URI validation belongs to the caller. pub(super) fn image_references(source: &str) -> Vec<(Range, String)> { + image_references_before(source, source.len(), false, 0, None) +} + +// Only search starts before the next competing token. Keep the complete source +// available because an image destination may itself contain Markdown markers. +fn image_references_before( + source: &str, + before: usize, + first_only: bool, + first_newline: usize, + matches: Option<&DelimiterMatches>, +) -> Vec<(Range, String)> { + // Fence segmentation takes priority even over image syntax that starts on + // an earlier line. Retain the document scanner for multiline helper input. + if first_only && first_newline < source.len() { + return image_references(source) + .into_iter() + .take(1) + .filter(|(range, _)| range.start < before) + .collect(); + } let mut images = Vec::new(); let mut fence = None; let mut prose_start = 0; let mut offset = 0; - for line in source.split_inclusive('\n') { + // A line without a fence needs no line-end scan before finding its first + // image. This common path also avoids repeatedly splitting a long line. + if first_only && before <= first_newline && fence_line(source).and_then(opening_fence).is_none() + { + collect_inline_images(source, 0, before, true, &mut images, matches); + return images; + } + for line in source[..before].split_inclusive('\n') { let raw = line.trim_end_matches(['\r', '\n']); let marker_line = fence_line(raw); if let Some((marker, minimum)) = fence { @@ -530,21 +644,53 @@ pub(super) fn image_references(source: &str) -> Vec<(Range, String)> { prose_start = offset + line.len(); } } else if let Some((marker, length, _)) = marker_line.and_then(opening_fence) { - collect_inline_images(&source[prose_start..offset], prose_start, &mut images); + collect_inline_images( + &source[prose_start..offset], + prose_start, + offset - prose_start, + first_only, + &mut images, + None, + ); + if first_only && !images.is_empty() { + return images; + } fence = Some((marker, length)); } offset += line.len(); } if fence.is_none() { - collect_inline_images(&source[prose_start..], prose_start, &mut images); + collect_inline_images( + &source[prose_start..], + prose_start, + before.saturating_sub(prose_start), + first_only, + &mut images, + None, + ); } images } -fn collect_inline_images(source: &str, base: usize, images: &mut Vec<(Range, String)>) { +fn collect_inline_images( + source: &str, + base: usize, + before: usize, + first_only: bool, + images: &mut Vec<(Range, String)>, + matches: Option<&DelimiterMatches>, +) { + let owned; + let matches = match matches { + Some(matches) => matches, + None => { + owned = DelimiterMatches::new(source); + &owned + } + }; let bytes = source.as_bytes(); let mut offset = 0; - while offset < bytes.len() { + while offset < before { match bytes[offset] { b'\\' if bytes.get(offset + 1).is_some_and(u8::is_ascii_punctuation) => { offset += 2; @@ -569,8 +715,11 @@ fn collect_inline_images(source: &str, base: usize, images: &mut Vec<(Range { - if let Some((end, destination)) = image_at(source, offset) { + if let Some((end, destination)) = image_at(source, offset, matches) { images.push((base + offset..base + end, destination)); + if first_only { + return; + } offset = end; } else { offset += 2; @@ -581,27 +730,9 @@ fn collect_inline_images(source: &str, base: usize, images: &mut Vec<(Range Option<(usize, String)> { +fn image_at(source: &str, start: usize, matches: &DelimiterMatches) -> Option<(usize, String)> { let bytes = source.as_bytes(); - let mut cursor = start + 2; - let mut depth = 1; - while cursor < bytes.len() { - match bytes[cursor] { - b'\\' if bytes.get(cursor + 1).is_some_and(u8::is_ascii_punctuation) => { - cursor += 2; - continue; - } - b'[' => depth += 1, - b']' => { - depth -= 1; - if depth == 0 { - break; - } - } - _ => {} - } - cursor += 1; - } + let mut cursor = matches.closing(source, start + 1, true)?; if bytes.get(cursor..cursor + 2)? != b"](" { return None; } @@ -612,7 +743,6 @@ fn image_at(source: &str, start: usize) -> Option<(usize, String)> { let angle = bytes.get(cursor) == Some(&b'<'); cursor += usize::from(angle); let destination_start = cursor; - depth = 0; while cursor < bytes.len() { match bytes[cursor] { b'\r' | b'\n' => return None, @@ -623,16 +753,21 @@ fn image_at(source: &str, start: usize) -> Option<(usize, String)> { b'>' if angle => break, b'"' | b'\'' if !angle - && depth == 0 && cursor > destination_start && bytes[cursor - 1].is_ascii_whitespace() => { break; } b'<' if angle => return None, - b'(' if !angle => depth += 1, - b')' if !angle && depth == 0 => break, - b')' if !angle => depth -= 1, + b'(' if !angle => { + // Inside nested parentheses quotes are destination text, not + // titles. Skip the balanced region, or fail immediately when + // it cannot close before a newline/end. Do not use this map + // for the outer delimiter: titles can contain unbalanced parens. + let base = matches.source_len - source.len(); + cursor = *matches.image_parentheses.get(&(base + cursor))? - base; + } + b')' if !angle => break, _ => {} } cursor += 1; @@ -679,35 +814,30 @@ fn image_at(source: &str, start: usize) -> Option<(usize, String)> { Some((cursor + 1, decoded)) } -fn next_markdown_link(source: &str) -> Option> { +fn next_markdown_link<'a>( + source: &'a str, + before: usize, + matches: &DelimiterMatches, +) -> Option> { let mut offset = 0; - while let Some(relative_start) = source[offset..].find('[') { + while offset < before { + let Some(relative_start) = source[offset..before].find('[') else { + break; + }; let start = offset + relative_start; // Images have their own rendering path; do not consume their alt text as a link. if start > 0 && source.as_bytes()[start - 1] == b'!' { - offset = image_at(source, start - 1).map_or(start + 1, |(end, _)| end); + offset = image_at(source, start - 1, matches).map_or(start + 1, |(end, _)| end); continue; } - let label_end = source[start + 1..].find(']').map(|end| start + 1 + end)?; + let base = matches.source_len - source.len(); + let label_end = *matches.raw_label_ends.get(&(base + start))? - base; if !source[label_end..].starts_with("](") { offset = label_end + 1; continue; } let url_start = label_end + 2; - let mut depth = 0; - let mut url_end = None; - for (relative, character) in source[url_start..].char_indices() { - match character { - '(' => depth += 1, - ')' if depth == 0 => { - url_end = Some(url_start + relative); - break; - } - ')' => depth -= 1, - _ => {} - } - } - let url_end = url_end?; + let url_end = matches.closing(source, url_start - 1, false)?; let url = &source[url_start..url_end]; if super::safe_media_uri(url) { return Some(Link { @@ -745,46 +875,47 @@ fn is_image_label(label: &str) -> bool { }) } -fn next_link(source: &str) -> Option> { - let bare = ["https://", "http://"] - .into_iter() - .filter_map(|scheme| source.find(scheme)) - .min() - .map(|start| { - let mut end = source[start..] - .find(char::is_whitespace) - .map_or(source.len(), |length| start + length); - loop { - let Some(character) = source[..end].chars().next_back() else { - break; - }; - let unmatched_close = character == ')' - && source[start..end].chars().filter(|&c| c == ')').count() - > source[start..end].chars().filter(|&c| c == '(').count(); - if matches!( - character, - '.' | ',' | ';' | ':' | '!' | '?' | ']' | '}' | '\'' | '"' - ) || unmatched_close - { - end -= character.len_utf8(); - } else { - break; - } - } - Link { - start, - end, - label: None, - url: &source[start..end], - url_range: start..end, - } +fn next_link<'a>( + source: &'a str, + before: usize, + bare_start: Option, + matches: &DelimiterMatches, +) -> Option> { + let bare_start = bare_start.filter(|start| *start < before); + if let Some(markdown) = next_markdown_link(source, bare_start.unwrap_or(before), matches) { + return Some(markdown); + } + bare_start.map(|start| { + let mut end = source[start..] + .find(char::is_whitespace) + .map_or(source.len(), |length| start + length); + let mut balance = source[start..end].bytes().fold(0isize, |balance, byte| { + balance + isize::from(byte == b')') - isize::from(byte == b'(') }); - let markdown = next_markdown_link(source); - match (bare, markdown) { - (Some(bare), Some(markdown)) if markdown.start < bare.start => Some(markdown), - (Some(bare), _) => Some(bare), - (None, markdown) => markdown, - } + loop { + let Some(character) = source[..end].chars().next_back() else { + break; + }; + let unmatched_close = character == ')' && balance > 0; + if matches!( + character, + '.' | ',' | ';' | ':' | '!' | '?' | ']' | '}' | '\'' | '"' + ) || unmatched_close + { + end -= character.len_utf8(); + balance -= isize::from(character == ')'); + } else { + break; + } + } + Link { + start, + end, + label: None, + url: &source[start..end], + url_range: start..end, + } + }) } pub(super) fn image_label_links(source: &str) -> Vec<(usize, String)> { @@ -838,12 +969,83 @@ fn inline_with_link_destinations_and_ranges( let mut spans = Vec::new(); let mut plain = String::new(); let mut rest = source; + let matches = DelimiterMatches::new(source); + let mut markers = source.match_indices(['`', '*', '_']).peekable(); + let mut schemes = source + .match_indices("http") + .filter_map(|(index, _)| { + (source[index..].starts_with("https://") || source[index..].starts_with("http://")) + .then_some(index) + }) + .peekable(); + let mut image_starts = source + .match_indices("![") + .map(|(index, _)| index) + .peekable(); + let mut newlines = source + .match_indices('\n') + .map(|(index, _)| index) + .peekable(); + // A failed underscore closer search must not rescan the remaining line for + // every opener. These are precisely the old parser's eligible closers. + let underscore_closers: Vec<_> = source + .match_indices('_') + .filter_map(|(index, _)| { + (!source[..index].ends_with(char::is_whitespace) + && source[index + 1..] + .chars() + .next() + .is_none_or(|character| !character.is_alphanumeric())) + .then_some(index) + }) + .collect(); while !rest.is_empty() { - let marker = rest - .find(['`', '*', '_']) - .map(|index| (index, &rest[index..])); - let image = image_references(rest).into_iter().next(); - let link = next_link(rest); + let relative = source.len() - rest.len(); + while markers.peek().is_some_and(|(index, _)| *index < relative) { + markers.next(); + } + let marker = markers.peek().map(|(index, _)| { + let index = *index - relative; + (index, &rest[index..]) + }); + let before = marker.as_ref().map_or(rest.len(), |(index, _)| *index); + while schemes.peek().is_some_and(|index| *index < relative) { + schemes.next(); + } + while newlines.peek().is_some_and(|index| *index < relative) { + newlines.next(); + } + while image_starts.peek().is_some_and(|index| *index < relative) { + image_starts.next(); + } + let image_start = image_starts + .peek() + .map_or(before, |index| (*index - relative).min(before)); + let bare_start = schemes.peek().map(|index| *index - relative); + let mut link = next_link(rest, image_start, bare_start, &matches); + let image = if link.is_none() && image_start < before { + image_references_before( + rest, + before, + true, + newlines + .peek() + .map_or(rest.len(), |index| *index - relative), + Some(&matches), + ) + .into_iter() + .next() + } else { + None + }; + if link.is_none() && image_start < before { + link = next_link( + rest, + image.as_ref().map_or(before, |(range, _)| range.start), + bare_start, + &matches, + ); + } let consumed = offset + source.len() - rest.len(); let label = labels .iter() @@ -946,14 +1148,10 @@ fn inline_with_link_destinations_and_ranges( // Emphasis needs the markers hugging the text, so arithmetic like // `2 * 3 * 4` stays arithmetic instead of turning italic. let paired = if delimiter == "_" { - body.match_indices('_').map(|(index, _)| index).find(|end| { - *end > 0 - && !body[..*end].ends_with(char::is_whitespace) - && body[*end + 1..] - .chars() - .next() - .is_none_or(|character| !character.is_alphanumeric()) - }) + let body_start = source.len() - body.len(); + underscore_closers + .get(underscore_closers.partition_point(|index| *index <= body_start)) + .map(|index| *index - body_start) } else { body.find(delimiter).filter(|end| { *end > 0 @@ -1037,6 +1235,257 @@ mod tests { .collect() } + #[test] + fn incomplete_streaming_delimiters_preserve_later_complete_tokens() { + for count in [32, 1000] { + let prefix = "![".repeat(count); + let source = format!("{prefix}![inner](local.png)"); + let images = image_references(&source); + assert_eq!(images, [(prefix.len()..source.len(), "local.png".into())]); + let spans = inline_spans(&source, Style::default()); + assert_eq!( + spans + .iter() + .map(|span| span.span.content.as_ref()) + .collect::(), + source + ); + assert_eq!( + spans + .iter() + .filter(|span| span.image_end) + .map(|span| span.span.content.as_ref()) + .collect::>(), + ["![inner](local.png)"] + ); + + let prefix = "[label](https://example.com/( ".repeat(count); + let source = format!("{prefix}[Image #7](file:///good)"); + let spans = inline_spans(&source, Style::default()); + assert_eq!( + spans + .iter() + .map(|span| span.span.content.as_ref()) + .collect::(), + format!("{prefix}Image #7") + ); + let urls = spans + .iter() + .filter_map(|span| span.url.as_deref()) + .collect::>(); + assert_eq!(urls.len(), count + 1); + assert!( + urls[..count] + .iter() + .all(|url| *url == "https://example.com/(") + ); + assert_eq!(urls[count], "file:///good"); + let destinations = image_label_link_destinations(&source); + let start = source.rfind("file:///good").unwrap(); + assert_eq!( + destinations, + [(start..start + "file:///good".len(), "file:///good".into())] + ); + } + } + + #[test] + fn repeated_rich_inline_tokens_preserve_styles_links_images_and_ranges() { + let unit = + "**bold _inner_** [Image #1](file:///tmp/a_(b).png) ![alt](a_b.png) `literal *code*` "; + let source = unit.repeat(1000); + let spans = inline_spans(&source, Style::default()); + let visible = spans + .iter() + .map(|span| span.span.content.as_ref()) + .collect::(); + assert_eq!( + visible, + "bold inner Image #1 ![alt](a_b.png) `literal *code*` ".repeat(1000) + ); + assert_eq!(spans.iter().filter(|span| span.image_end).count(), 1000); + let inner = spans + .iter() + .filter(|span| span.span.content == "inner") + .collect::>(); + assert_eq!(inner.len(), 1000); + assert!(inner.iter().all(|span| { + span.span + .style + .add_modifier + .contains(Modifier::BOLD | Modifier::ITALIC) + })); + let destinations = image_label_link_destinations(&source); + assert_eq!(destinations.len(), 1000); + for (index, (range, url)) in destinations.iter().enumerate() { + assert_eq!(&source[range.clone()], "file:///tmp/a_(b).png"); + assert_eq!(url, "file:///tmp/a_(b).png"); + assert_eq!( + range.start, + index * unit.len() + unit.find("file:///").unwrap() + ); + } + } + + #[test] + fn bounded_searches_preserve_malformed_syntax_and_url_trimming() { + for source in [ + "_open ".repeat(1000), + "![".repeat(1000), + "[unfinished".repeat(1000), + ] { + let spans = inline_spans(&source, Style::default()); + assert_eq!( + spans + .iter() + .map(|span| span.span.content.as_ref()) + .collect::(), + source + ); + assert!( + spans + .iter() + .all(|span| span.url.is_none() && !span.image_end) + ); + } + let suffix = ")".repeat(5000) + "...!?"; + let source = format!("https://example.com/a_(b){suffix}"); + let spans = inline_spans(&source, Style::default()); + assert_eq!(spans[0].url.as_deref(), Some("https://example.com/a_(b)")); + assert_eq!( + spans + .iter() + .map(|span| span.span.content.as_ref()) + .collect::(), + source + ); + let source = r"\![escaped](local) **[Image #1](file:///a)** `![code](b)` ![real](c)"; + let spans = inline_spans(source, Style::default()); + assert_eq!( + spans + .iter() + .filter(|span| span.image_end) + .map(|span| span.span.content.as_ref()) + .collect::>(), + ["![real](c)"] + ); + let ranges = image_label_link_destinations(source); + assert_eq!(ranges.len(), 1); + assert_eq!(&source[ranges[0].0.clone()], "file:///a"); + } + + #[test] + fn multiline_fences_still_bound_images_before_inline_markers() { + let source = "]![![`~~~\\[Image #1](file:///a)[Image #1](file:///a)file:///a\n~~~](~~~***)"; + let spans = inline_spans(source, Style::default()); + assert_eq!( + spans + .iter() + .filter(|span| span.image_end) + .map(|span| span.span.content.as_ref()) + .collect::>(), + ["![`~~~\\[Image #1](file:///a)"] + ); + let ranges = image_label_link_destinations(source); + assert_eq!(ranges.len(), 1); + assert_eq!(&source[ranges[0].0.clone()], "file:///a"); + } + + #[test] + fn incomplete_image_destinations_and_raw_labels_keep_later_tokens() { + for count in [1, 64, 1024] { + let prefix = "![x](".repeat(count); + assert!(image_references(&prefix).is_empty()); + let source = format!("{prefix}![雪](ok.png) [docs](https://example.com)"); + assert_eq!( + image_references(&source), + vec![( + prefix.len()..prefix.len() + "![雪](ok.png)".len(), + "ok.png".into() + )] + ); + let spans = inline_spans(&source, Style::default()); + assert!(spans.iter().any(|span| span.image_end)); + assert!( + spans + .iter() + .any(|span| span.url.as_deref() == Some("https://example.com")) + ); + + let prefix = "[* ".repeat(count); + assert!( + next_markdown_link(&prefix, prefix.len(), &DelimiterMatches::new(&prefix)) + .is_none() + ); + let source = format!("{prefix}雪](https://example.com)"); + let link = + next_markdown_link(&source, source.len(), &DelimiterMatches::new(&source)).unwrap(); + assert_eq!(link.label, Some(&source[1..source.find(']').unwrap()])); + assert_eq!(link.url, "https://example.com"); + } + // Link labels intentionally stop at the first raw ], even when escaped. + let source = r"[雪\](https://example.com)"; + let link = + next_markdown_link(source, source.len(), &DelimiterMatches::new(source)).unwrap(); + assert_eq!(link.label, Some(r"雪\")); + } + + #[test] + fn image_destination_summary_preserves_escape_title_and_angle_rules() { + for (source, destination) in [ + (r#"![雪](a(b(c)).png "unbalanced ( title")"#, "a(b(c)).png"), + ( + r#"![雪](a(b "literal (c)").png 'title )')"#, + "a(b \"literal (c)\").png", + ), + (r#"![雪](a\(b\).png "escaped \" ) title")"#, "a(b).png"), + (r#"![雪](a(b\)c).png 'escaped \' ( title')"#, "a(b)c).png"), + (r#"![雪](<雪(a> "title ( )")"#, "雪(a"), + ] { + let images = image_references(source); + assert_eq!( + images, + vec![(0..source.len(), destination.into())], + "{source}" + ); + } + for source in ["![x](a(b\nc))", "![x](a(b\rc))", r"![x](a(b\))"] { + assert!(image_references(source).is_empty(), "{source}"); + } + let source = "![x](a(b\n![雪](ok.png)\n```\n![hidden](no)\n```"; + let images = image_references(source); + assert_eq!(images.len(), 1); + assert_eq!(images[0].1, "ok.png"); + } + + /// Run manually with `mise run test -- --release --lib inline_scaling_probe -- --ignored --nocapture`. + #[test] + #[ignore = "manual timing probe; reports measurements without timing assertions"] + fn inline_scaling_probe() { + use std::{hint::black_box, time::Instant}; + for (name, unit) in [ + ("emphasis", "*bold* plain "), + ("nested", "**bold _inner_** plain "), + ("links", "[label](https://example.com/a_(b)) "), + ("images", "![alt](https://example.com/a_(b).png) "), + ("bare", "https://example.com/a_(b))))... "), + ("underscore", "_open "), + ("unclosed_images", "!["), + ("image_destinations", "![x]("), + ("raw_labels", "[* "), + ("unclosed_links", "[label](https://example.com/( "), + ] { + for count in [1000, 5000, 10000] { + let source = unit.repeat(count); + let start = Instant::now(); + let spans = inline_spans(black_box(&source), Style::default()); + let elapsed = start.elapsed(); + eprintln!("{name:18} {count:6} {:9} bytes {elapsed:?}", source.len()); + black_box(spans); + } + } + } + #[test] fn trusted_image_labels_preserve_surrounding_emphasis_and_reject_forged_uris() { let text = @@ -1431,9 +1880,17 @@ mod tests { #[test] fn markdown_link_parser_does_not_consume_images() { - assert!(next_markdown_link("![alt](https://example.com/image.png)").is_none()); + assert!( + next_markdown_link( + "![alt](https://example.com/image.png)", + "![alt](https://example.com/image.png)".len(), + &DelimiterMatches::new("![alt](https://example.com/image.png)") + ) + .is_none() + ); let source = "![alt](https://example.com/image.png) [docs](https://example.com/docs)"; - let link = next_markdown_link(source).unwrap(); + let link = + next_markdown_link(source, source.len(), &DelimiterMatches::new(source)).unwrap(); assert_eq!(link.label, Some("docs")); assert_eq!( &source[link.start..link.end], diff --git a/src/tui/mod.rs b/src/tui/mod.rs index f2b7df3..d457e94 100644 --- a/src/tui/mod.rs +++ b/src/tui/mod.rs @@ -15,6 +15,7 @@ mod image; #[cfg(all(test, unix))] mod keyboard_tests; mod markdown; +mod scheduler; mod source; mod startup; @@ -1850,6 +1851,7 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( Background(Option), Update(Option), Tick, + Frame, Stop, } let mut next_priority = 0; @@ -1857,6 +1859,7 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( let mut pending_update = None; let mut clipboard_pastes = ClipboardPastes::default(); let mut submit_after_paste = false; + let mut frames = scheduler::Frames::new(tokio::time::Instant::now()); loop { // Reconcile after every prior event, including failures in // result/observe, before accepting any next user input. @@ -1866,18 +1869,18 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( &app.clipboard_route(), ); if let Some(queued) = pending_update.take() { - match background_workers.try_update(queued) { - Ok(()) => {} - Err(error) => match *error { - std::sync::mpsc::TrySendError::Full(queued) => { - pending_update = Some(queued); - } - std::sync::mpsc::TrySendError::Disconnected(_) => return Ok(()), - }, - } + pending_update = match scheduler::forward_updates( + &background_workers, &mut updates_rx, queued, + ) { + Ok(pending) => pending, + Err(()) => return Ok(()), + }; + } + if frames.ready(tokio::time::Instant::now()) { + draw_frame(&mut terminal, &mut app, &mut images) + .map_err(agent_client_protocol::Error::into_internal_error)?; + frames.drawn(tokio::time::Instant::now()); } - draw_frame(&mut terminal, &mut app, &mut images) - .map_err(agent_client_protocol::Error::into_internal_error)?; let event = if std::mem::take(&mut submit_after_paste) { SessionEvent::Terminal(Some(Ok(Event::Key(KeyEvent::new( KeyCode::Enter, @@ -1885,6 +1888,10 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( ))))) } else { let redraw = app.needs_redraw_tick() || images.pending(); + let frame_deadline = frames.deadline(); + let mut frame = pin!(tokio::time::sleep_until( + frame_deadline.unwrap_or_else(tokio::time::Instant::now), + )); let mut stopped = pin!(stop.requested()); let mut shutdown = pin!(storage_shutdown.cancelled()); // A local round-robin race keeps hot input/update queues @@ -1895,7 +1902,7 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( // until this wait ends. This scope drops all losers and // releases input borrows before handlers drain/reset them. poll_fn(|cx| { - let sources = 8; + let sources = 9; for offset in 0..sources { let branch = (next_priority + offset) % sources; let ready = match branch { @@ -1917,6 +1924,7 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( 6 => stopped.as_mut().poll(cx).map(|()| SessionEvent::Stop), 7 => voice.poll(cx).map(SessionEvent::Voice), + 8 if frame_deadline.is_some() => frame.as_mut().poll(cx).map(|()| SessionEvent::Frame), _ => Poll::Pending, }; if ready.is_ready() { @@ -1927,6 +1935,12 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( Poll::Pending }).await }; + // Invalidate before handlers: early `continue`s can also mutate + // visible state. Worker forwarding alone never requests a frame. + if matches!(&event, SessionEvent::Terminal(_) | SessionEvent::Voice(_) + | SessionEvent::ModelSwitch(_) | SessionEvent::Tick) { + frames.invalidate(); + } match event { SessionEvent::Voice(event) => match event { @@ -1989,12 +2003,12 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( // instead of one frame per character. let mut next = terminal_event; let mut action = Action::None; - for _ in 0..MAX_BURST { + for index in 0..MAX_BURST { match next { Some(Ok(event)) => action = handle_with_clipboard(&mut app, &mut clipboard_pastes, event), Some(Err(_)) | None => return Ok(()), } - if !matches!(action, Action::None) { + if !matches!(action, Action::None) || index + 1 == MAX_BURST { break; } // `EventStream::next().now_or_never()` polls with a noop @@ -2540,25 +2554,15 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( } }, SessionEvent::Background(completion) => match completion { - Some(BackgroundCompletion::Update { queued, images }) => { - if let Some(update) = accept_queued_update(&transition_session, queued) { - if let Update::ConfigOptions(options) = &update { - refresh_config_state(&mut app, Some(options)); - } - - voice.observe(&update, &mut app); - app.apply_materialized(update, images); - } - } - Some(BackgroundCompletion::Clipboard { generation, route, result }) => { - submit_after_paste = finish_clipboard_paste( - &mut app, - &transition_session, - &mut clipboard_pastes, - generation, - route, - result, + Some(completion) => { + let applied = scheduler::apply_completions( + &mut app, &transition_session, &mut voice, + &mut clipboard_pastes, &mut background_rx, completion, ); + if applied.dirty { + frames.invalidate(); + } + submit_after_paste = applied.submit_after_paste; } None => return Ok(()), }, @@ -2594,17 +2598,17 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( } }, SessionEvent::Update(update) => match update { - Some(update) => match background_workers.try_update(update) { - Ok(()) => {} - Err(error) => match *error { - std::sync::mpsc::TrySendError::Full(update) => { - pending_update = Some(update); - } - std::sync::mpsc::TrySendError::Disconnected(_) => return Ok(()), - }, - }, + Some(update) => { + pending_update = match scheduler::forward_updates( + &background_workers, &mut updates_rx, update, + ) { + Ok(pending) => pending, + Err(()) => return Ok(()), + }; + } None => return Ok(()), }, + SessionEvent::Frame => {}, SessionEvent::Tick => app.tick(), SessionEvent::Stop => return Ok(()), } diff --git a/src/tui/scheduler.rs b/src/tui/scheduler.rs new file mode 100644 index 0000000..80257ee --- /dev/null +++ b/src/tui/scheduler.rs @@ -0,0 +1,446 @@ +//! Bounded queue turns and event-independent frame pacing for the active session. + +use super::{ + ActiveSessionRoute, App, BackgroundCompletion, BackgroundWorkers, ClipboardPastes, NativeVoice, + QueuedUpdate, Update, accept_queued_update, finish_clipboard_paste, refresh_config_state, +}; +use std::sync::{Arc, Mutex}; +use tokio::{sync::mpsc, time::Instant}; + +const FRAME_INTERVAL: std::time::Duration = std::time::Duration::from_millis(16); +// Yield to the round-robin event selector even when producers keep queues hot. +const UPDATE_BURST: usize = 64; + +pub(super) struct Frames { + dirty: bool, + next: Instant, +} + +impl Frames { + pub(super) fn new(now: Instant) -> Self { + Self { + dirty: true, + next: now, + } + } + + pub(super) fn invalidate(&mut self) { + self.dirty = true; + } + + pub(super) fn deadline(&self) -> Option { + self.dirty.then_some(self.next) + } + + pub(super) fn ready(&self, now: Instant) -> bool { + self.dirty && now >= self.next + } + + pub(super) fn drawn(&mut self, now: Instant) { + self.dirty = false; + // Do not catch up missed frames after slow draws or blocking actions. + self.next = now + FRAME_INTERVAL; + } +} + +/// Keep a full worker queue's head ahead of all later updates. No update bypasses +/// image materialization, including non-image updates following an image. +pub(super) fn forward_updates( + workers: &BackgroundWorkers, + updates: &mut mpsc::UnboundedReceiver, + first: QueuedUpdate, +) -> Result, ()> { + for queued in std::iter::once(first) + .chain(std::iter::from_fn(|| updates.try_recv().ok())) + .take(UPDATE_BURST) + { + match workers.try_update(queued) { + Ok(()) => {} + Err(error) => match *error { + std::sync::mpsc::TrySendError::Full(queued) => return Ok(Some(queued)), + std::sync::mpsc::TrySendError::Disconnected(_) => return Err(()), + }, + } + } + Ok(None) +} + +pub(super) struct Applied { + pub(super) dirty: bool, + pub(super) submit_after_paste: bool, +} + +/// Apply completions in FIFO order and coalesce their redraw requests. Keep +/// voice/config observation on every accepted update, never just the last one. +pub(super) fn apply_completions( + app: &mut App, + route: &Arc>, + voice: &mut NativeVoice, + pastes: &mut ClipboardPastes, + completed: &mut mpsc::Receiver, + first: BackgroundCompletion, +) -> Applied { + let mut applied = Applied { + dirty: false, + submit_after_paste: false, + }; + for completion in std::iter::once(first) + .chain(std::iter::from_fn(|| completed.try_recv().ok())) + .take(UPDATE_BURST) + { + match completion { + BackgroundCompletion::Update { queued, images } => { + if let Some(update) = accept_queued_update(route, queued) { + if let Update::ConfigOptions(options) = &update { + refresh_config_state(app, Some(options)); + } + voice.observe(&update, app); + app.apply_materialized(update, images); + applied.dirty = true; + } + } + BackgroundCompletion::Clipboard { + generation, + route: clipboard_route, + result, + } => { + applied.submit_after_paste = + finish_clipboard_paste(app, route, pastes, generation, clipboard_route, result); + applied.dirty = true; + // The synthetic Enter must precede the next completion, just as + // it did when the active loop consumed one completion per turn. + if applied.submit_after_paste { + break; + } + } + } + } + applied +} + +#[cfg(test)] +#[allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::unreachable, + clippy::disallowed_methods, + clippy::disallowed_macros +)] +mod tests { + use super::*; + use crate::tui::{ClipboardResult, VoiceHandoff, app::Block}; + use std::sync::atomic::AtomicBool; + + #[test] + fn dirty_frames_coalesce_without_postponing_the_deadline() { + let now = Instant::now(); + let mut frames = Frames::new(now); + assert!(frames.ready(now)); + frames.drawn(now); + assert_eq!(frames.deadline(), None); + assert!(!frames.ready(now + FRAME_INTERVAL)); + frames.invalidate(); + let deadline = frames.deadline().unwrap(); + frames.invalidate(); + assert_eq!(frames.deadline(), Some(deadline)); + assert!(!frames.ready(now)); + assert!(frames.ready(deadline)); + // A late frame schedules from its completion, not from the old deadline. + let late = deadline + FRAME_INTERVAL; + frames.drawn(late); + frames.invalidate(); + assert!(!frames.ready(late)); + assert!(frames.ready(late + FRAME_INTERVAL)); + } + + #[test] + fn forwarding_preserves_order_across_worker_backpressure() { + let (sender, worker) = std::sync::mpsc::sync_channel(1); + let workers = BackgroundWorkers { + updates: Some(sender), + clipboard: None, + stopping: Arc::new(AtomicBool::new(false)), + threads: vec![], + }; + let (sender, mut updates) = mpsc::unbounded_channel(); + for text in ["one", "two", "three"] { + sender + .send(QueuedUpdate::for_session(7, Update::Log(text.into()))) + .unwrap(); + } + let first = updates.try_recv().unwrap(); + let mut pending = forward_updates(&workers, &mut updates, first).unwrap(); + let mut received = vec![]; + loop { + let queued = worker.try_recv().unwrap(); + assert_eq!(queued.generation, Some(7)); + let Update::Log(text) = queued.update else { + panic!("expected log") + }; + received.push(text); + let Some(first) = pending else { break }; + pending = forward_updates(&workers, &mut updates, first).unwrap(); + } + assert_eq!(received, ["one", "two", "three"]); + assert!(updates.try_recv().is_err()); + drop(worker); + assert!( + forward_updates( + &workers, + &mut updates, + QueuedUpdate::global(Update::Log("closed".into())) + ) + .is_err() + ); + } + + #[test] + fn forwarding_yields_with_capacity_and_never_skips_the_next_head() { + let total = UPDATE_BURST * 2; + let (sender, worker) = std::sync::mpsc::sync_channel(total); + let workers = BackgroundWorkers { + updates: Some(sender), + clipboard: None, + stopping: Arc::new(AtomicBool::new(false)), + threads: vec![], + }; + let (sender, mut updates) = mpsc::unbounded_channel(); + for index in 0..total { + sender + .send(QueuedUpdate::global(Update::Log(index.to_string()))) + .unwrap(); + } + let first = updates.try_recv().unwrap(); + assert!( + forward_updates(&workers, &mut updates, first) + .unwrap() + .is_none() + ); + let next = updates + .try_recv() + .expect("a bounded turn leaves queued work"); + assert!( + forward_updates(&workers, &mut updates, next) + .unwrap() + .is_none() + ); + let received: Vec<_> = worker + .try_iter() + .map(|queued| { + let Update::Log(text) = queued.update else { + panic!("expected log") + }; + text + }) + .collect(); + assert_eq!( + received, + (0..total) + .map(|index| index.to_string()) + .collect::>() + ); + assert!(updates.try_recv().is_err()); + } + + #[test] + fn stale_clipboard_completion_cannot_finish_current_paste() { + let mut app = app(); + let clipboard_route = app.clipboard_route(); + let mut pastes = ClipboardPastes::default(); + pastes.queued(7, clipboard_route.clone()); + let (sender, mut completed) = mpsc::channel(2); + sender + .try_send(BackgroundCompletion::Clipboard { + generation: 7, + route: clipboard_route.clone(), + result: ClipboardResult::Text("current".into()), + }) + .ok() + .unwrap(); + let applied = apply_completions( + &mut app, + &route(), + &mut NativeVoice::default(), + &mut pastes, + &mut completed, + BackgroundCompletion::Clipboard { + generation: 6, + route: clipboard_route, + result: ClipboardResult::Text("stale".into()), + }, + ); + assert!(applied.dirty); + assert!(!applied.submit_after_paste); + assert_eq!(app.editor.text(), "current"); + assert!(pastes.pending.is_none()); + } + + fn app() -> App { + App::new( + "/tmp".into(), + "provider".into(), + "model".into(), + "a2a".into(), + ) + } + + fn route() -> Arc> { + Arc::new(Mutex::new(ActiveSessionRoute { + id: "session".into(), + generation: 7, + })) + } + + fn completion(generation: u64, update: Update) -> BackgroundCompletion { + BackgroundCompletion::Update { + queued: QueuedUpdate::for_session(generation, update), + images: vec![], + } + } + + #[test] + fn completions_apply_ordered_replacements_and_observe_every_voice_update() { + let mut app = app(); + let mut voice = NativeVoice::default(); + voice.handoff = Some(VoiceHandoff { + id: "task".into(), + accepted: false, + user_message_id: None, + text: String::new(), + message_id: String::new(), + }); + let (sender, mut completed) = mpsc::channel(16); + let updates = [ + Update::VoicePromptAccepted { + id: "task".into(), + result: Ok(()), + }, + Update::UserMessage { + id: "user".into(), + text: "question".into(), + images: vec![], + append: false, + }, + Update::AgentMessage { + id: "answer".into(), + text: "old".into(), + append: false, + }, + Update::AgentMessage { + id: "answer".into(), + text: "new".into(), + append: false, + }, + Update::AgentMessage { + id: "answer".into(), + text: " answer".into(), + append: true, + }, + ]; + for update in updates { + sender.try_send(completion(7, update)).ok().unwrap(); + } + // Stale data must not append to either the app or voice's accepted turn. + sender + .try_send(completion( + 6, + Update::AgentMessage { + id: "answer".into(), + text: " stale".into(), + append: true, + }, + )) + .ok() + .unwrap(); + let first = completed.try_recv().unwrap(); + let applied = apply_completions( + &mut app, + &route(), + &mut voice, + &mut ClipboardPastes::default(), + &mut completed, + first, + ); + assert!(applied.dirty); + assert!(!applied.submit_after_paste); + assert!(completed.try_recv().is_err()); + let handoff = voice.handoff.as_ref().unwrap(); + assert!(handoff.accepted); + assert_eq!(handoff.user_message_id.as_deref(), Some("user")); + assert_eq!(handoff.text, "new answer"); + assert!(matches!(app.blocks.last(), Some(Block::Agent(text)) if text == "new answer")); + } + + #[test] + fn stale_completions_do_not_request_a_frame() { + let mut app = app(); + let (_sender, mut completed) = mpsc::channel(1); + let applied = apply_completions( + &mut app, + &route(), + &mut NativeVoice::default(), + &mut ClipboardPastes::default(), + &mut completed, + completion(6, Update::Log("stale".into())), + ); + assert!(!applied.dirty); + assert!(app.blocks.is_empty()); + } + + #[test] + fn clipboard_submission_is_a_batch_boundary() { + let mut app = app(); + let clipboard_route = app.clipboard_route(); + let mut pastes = ClipboardPastes::default(); + pastes.queued(7, clipboard_route.clone()); + pastes.pending.as_mut().unwrap().submit = true; + let (sender, mut completed) = mpsc::channel(2); + sender + .try_send(completion(7, Update::Log("after paste".into()))) + .ok() + .unwrap(); + let applied = apply_completions( + &mut app, + &route(), + &mut NativeVoice::default(), + &mut pastes, + &mut completed, + BackgroundCompletion::Clipboard { + generation: 7, + route: clipboard_route, + result: ClipboardResult::Text("pasted".into()), + }, + ); + assert!(applied.dirty); + assert!(applied.submit_after_paste); + assert_eq!(app.editor.text(), "pasted"); + assert!(completed.try_recv().is_ok()); + assert!(app.blocks.is_empty()); + } + + #[test] + fn completion_turn_leaves_backlog_for_other_event_sources() { + let mut app = app(); + let (sender, mut completed) = mpsc::channel(UPDATE_BURST * 2); + for _ in 0..UPDATE_BURST * 2 { + sender + .try_send(completion(7, Update::Log("queued".into()))) + .ok() + .unwrap(); + } + let first = completed.try_recv().unwrap(); + let applied = apply_completions( + &mut app, + &route(), + &mut NativeVoice::default(), + &mut ClipboardPastes::default(), + &mut completed, + first, + ); + assert!(applied.dirty); + assert!(!app.logs.is_empty()); + assert!(app.logs.iter().all(|log| log == "queued")); + assert!(completed.try_recv().is_ok()); + } +} diff --git a/src/tui/ui.rs b/src/tui/ui.rs index 93033ff..c291e1b 100644 --- a/src/tui/ui.rs +++ b/src/tui/ui.rs @@ -1018,6 +1018,7 @@ fn draw_transcript(frame: &mut Frame<'_>, app: &mut App, images: &mut ImageRunti } let width = inner.width.max(1) as usize; + app.viewport = inner.height as usize; refresh_transcript_cache_with_images(app, images, width); let working_rows = if app.working() { wrap_linked_tagged( @@ -1246,6 +1247,19 @@ fn welcome_logo() -> Paragraph<'static> { /// Renders the transcript, tagging each line with the tool call it belongs to /// so a click on a card can be traced back to it. fn refresh_transcript_cache_with_images(app: &mut App, images: &mut ImageRuntime, width: usize) { + // Resolve content/width changes before choosing the animated viewport. + // Otherwise a large append or replacement can move follow-bottom to cards + // outside the old viewport and leave them stale for a frame. + let refreshed = refresh_transcript_cache_pass(app, images, width, None); + refresh_transcript_cache_pass(app, images, width, Some(&refreshed)); +} + +fn refresh_transcript_cache_pass( + app: &mut App, + images: &mut ImageRuntime, + width: usize, + refreshed_content: Option<&[usize]>, +) -> Vec { let structure_changed = app.transcript_revisions.len() != app.blocks.len() || app.transcript_cache.len() != app.blocks.len() || app.transcript_prefixes.len() != app.blocks.len() + 1; @@ -1258,16 +1272,37 @@ fn refresh_transcript_cache_with_images(app: &mut App, images: &mut ImageRuntime app.transcript_cache_width = width; app.transcript_dirty.extend(0..app.blocks.len()); } - app.transcript_dirty - .extend(app.transcript_dynamic.iter().copied()); - let animated_owners: std::collections::BTreeSet<_> = app - .transcript_dynamic - .iter() - .filter_map(|index| app.tool_owners.get(index).copied()) - .collect(); + // Content changes are always refreshed, but animation alone must not + // rebuild every running card in a long, scrolled-away transcript. Include + // a viewport of slack for working rows and small timer-wrap changes. + let total = app.transcript_prefixes.last().copied().unwrap_or(0); + let viewport = app.viewport.max(1); + let top = if app.follow { + total.saturating_sub(viewport) + } else { + app.scroll + }; + let animated_owners: std::collections::BTreeSet<_> = if let Some(refreshed) = refreshed_content + { + app.transcript_dynamic + .iter() + .map(|index| app.tool_owners.get(index).copied().unwrap_or(*index)) + .filter(|&index| { + refreshed.binary_search(&index).is_err() + && app.transcript_prefixes[index + 1] > top.saturating_sub(viewport) + && app.transcript_prefixes[index] < top.saturating_add(viewport * 2) + }) + .collect() + } else { + std::collections::BTreeSet::new() + }; app.transcript_dirty.extend(animated_owners.iter().copied()); let dirty = std::mem::take(&mut app.transcript_dirty); let mut first_changed_count = app.blocks.len(); + let mut last_changed_count = 0; + // Dirty indices are ordered, so this remains sorted for cheap membership + // checks in the animation pass without allocating another tree. + let mut refreshed = Vec::new(); for block_index in dirty { let dynamic = match &app.blocks[block_index] { Block::Thought { millis, .. } => millis.is_none(), @@ -1288,17 +1323,47 @@ fn refresh_transcript_cache_with_images(app: &mut App, images: &mut ImageRuntime let old_count = app.transcript_cache[block_index] .as_ref() .map_or(0, |cached| cached.rows.len()); - let (rows, cached_images) = - transcript_block_rows(app, block_index, width, images.enabled()); - if missing || rows.len() != old_count { + let text = match &app.blocks[block_index] { + Block::Agent(text) => Some(text.as_str()), + Block::AgentParts(parts) => match parts.as_slice() { + [AgentPart::Text(text)] => Some(text.as_str()), + _ => None, + }, + _ => None, + }; + let cached = if let Some(text) = text { + let previous = app.transcript_cache[block_index] + .take() + .filter(|_| !width_changed); + incremental_agent_rows( + text, + block_index, + width, + images.enabled(), + revision, + previous, + ) + } else { + let (rows, cached_images) = + transcript_block_rows(app, block_index, width, images.enabled()); + CachedTranscriptBlock { + revision, + rows, + images: cached_images, + stable_prefix: String::new(), + prefix_rows: 0, + prefix_lines: 0, + } + }; + if missing || cached.rows.len() != old_count { first_changed_count = first_changed_count.min(block_index); + last_changed_count = block_index; layout_changed |= !missing; } - app.transcript_cache[block_index] = Some(CachedTranscriptBlock { - revision, - rows, - images: cached_images, - }); + app.transcript_cache[block_index] = Some(cached); + if refreshed_content.is_none() { + refreshed.push(block_index); + } if dynamic { app.transcript_dynamic.insert(block_index); } else { @@ -1309,13 +1374,79 @@ fn refresh_transcript_cache_with_images(app: &mut App, images: &mut ImageRuntime let rows = app.transcript_cache[index] .as_ref() .map_or(0, |cached| cached.rows.len()); - app.transcript_prefixes[index + 1] = app.transcript_prefixes[index] + let next = app.transcript_prefixes[index] + rows + usize::from(rows > 0 && app.transcript_prefixes[index] > 0); + let unchanged = next == app.transcript_prefixes[index + 1]; + app.transcript_prefixes[index + 1] = next; + if index >= last_changed_count && unchanged { + break; + } } if layout_changed { app.clear_transcript_interaction(); } + refreshed +} + +/// Cache only complete, line-local source lines. The first potentially structural +/// line and everything after it remain one renderer input: table lookahead, +/// unfinished fences (including copy ranges), links and images retain the full +/// renderer's semantics. Replacements and width changes safely start over. +fn incremental_agent_rows( + text: &str, + block: usize, + width: usize, + reserve_images: bool, + revision: u64, + previous: Option, +) -> CachedTranscriptBlock { + let mut cached = previous + .filter(|cached| text.starts_with(&cached.stable_prefix)) + .unwrap_or_else(|| CachedTranscriptBlock { + revision, + rows: Vec::new(), + images: Vec::new(), + stable_prefix: String::new(), + prefix_rows: 0, + prefix_lines: 0, + }); + let offset = cached.stable_prefix.len(); + let suffix = &text[offset..]; + let mut stable_bytes = 0; + let mut stable_lines = 0; + for line in suffix.split_inclusive('\n') { + if !line.ends_with('\n') || line.contains(['`', '~', '|', '[']) { + break; + } + stable_bytes += line.len(); + stable_lines += 1; + } + cached.rows.truncate(cached.prefix_rows); + let (mut rows, mut images) = agent_block_rows(suffix, block, width, reserve_images); + let new_prefix_rows = rows + .iter() + .take_while(|row| row.1.2.is_some_and(|line| line < stable_lines)) + .count(); + for row in &mut rows { + if let Some(hit) = &mut row.1.1 { + hit.range.start += offset; + hit.range.end += offset; + } + if let Some(line) = &mut row.1.2 { + *line += cached.prefix_lines; + } + } + for image in &mut images { + image.row += cached.rows.len(); + } + cached.rows.extend(rows); + cached.images = images; + cached.stable_prefix.push_str(&suffix[..stable_bytes]); + cached.prefix_rows += new_prefix_rows; + cached.prefix_lines += stable_lines; + cached.revision = revision; + cached } fn user_block_rows(message: &UserMessage, width: usize) -> Vec { @@ -3144,6 +3275,304 @@ mod tests { .collect() } + fn assert_transcript_rows_equal( + actual: &[super::CachedTranscriptRow], + expected: &[super::CachedTranscriptRow], + ) { + assert_eq!(actual.len(), expected.len()); + for (actual, expected) in actual.iter().zip(expected) { + assert_eq!(actual.0, expected.0); + assert_eq!(actual.1.0, expected.1.0); + assert_eq!( + actual.1.1.as_ref().map(|hit| (hit.block, &hit.range)), + expected.1.1.as_ref().map(|hit| (hit.block, &hit.range)) + ); + assert_eq!(actual.1.2, expected.1.2); + assert_eq!(actual.2, expected.2); + assert_eq!(actual.3, expected.3); + } + } + + #[test] + #[ignore = "manual transcript history and streamed append performance probe"] + fn transcript_history_streaming_perf_probe() { + let mut app = App::new( + PathBuf::from("/tmp/kit"), + "test".into(), + "test".into(), + "0:0".into(), + ); + let mut images = ImageRuntime::disabled(); + for index in 0..10_000 { + app.apply(Update::AgentMessage { + id: index.to_string(), + text: "Historical response.\nSecond line.".into(), + append: false, + }); + } + let start = std::time::Instant::now(); + refresh_transcript_cache_with_images(&mut app, &mut images, 80); + eprintln!("history layout: {:?}", start.elapsed()); + for size in [100, 1_000] { + for (name, chunk) in [ + ("plain", "Another complete streamed line.\n"), + ("markdown-first", "# Heading\n- **formatted text**\n"), + ("unfinished-long-line", "another word "), + ("fenced-first-fallback", "```rust\nlet value = 1;\n"), + ] { + let mut times = Vec::new(); + for full in [false, true] { + let id = format!("probe-{name}-{size}-{full}"); + app.apply(Update::AgentMessage { + id: id.clone(), + text: chunk.repeat(size), + append: false, + }); + refresh_transcript_cache_with_images(&mut app, &mut images, 80); + let index = app.blocks.len() - 1; + let start = std::time::Instant::now(); + for _ in 0..100 { + app.apply(Update::AgentMessage { + id: id.clone(), + text: chunk.into(), + append: true, + }); + if full { + // Use the production full renderer on the same warmed source. + std::hint::black_box(super::transcript_block_rows( + &app, index, 80, false, + )); + } else { + refresh_transcript_cache_with_images(&mut app, &mut images, 80); + } + } + times.push(start.elapsed()); + } + eprintln!( + "warm append {name} initial_chunks={size}: cached={:?}, full={:?}", + times[0], times[1] + ); + } + } + let start = std::time::Instant::now(); + refresh_transcript_cache_with_images(&mut app, &mut images, 40); + eprintln!("history resize: {:?}", start.elapsed()); + for index in 0..100 { + app.apply(Update::ToolStarted { + id: format!("animated-{index}"), + title: "compose".into(), + kind: ToolKind::Other, + script: Some("value = 1\n".repeat(30)), + backgrounded: true, + }); + } + app.viewport = 24; + refresh_transcript_cache_with_images(&mut app, &mut images, 80); + for dirty in [false, true] { + let start = std::time::Instant::now(); + for _ in 0..100 { + if dirty { + app.apply(Update::ToolPatched { + id: "animated-99".into(), + title: None, + kind: None, + status: None, + script: None, + output: Some(vec!["more output".into()]), + images: None, + append_output: true, + intent: None, + backgrounded: false, + }); + } + refresh_transcript_cache_with_images(&mut app, &mut images, 80); + } + eprintln!( + "100 running cards, 100 draws, visible_content_dirty={dirty}: {:?}", + start.elapsed() + ); + } + } + + #[test] + fn dynamic_cards_refresh_on_first_draw_scroll_and_follow_after_replacement() { + let mut app = App::new( + PathBuf::from("/tmp/kit"), + "test".into(), + "test".into(), + "0:0".into(), + ); + for (id, title) in [("owner", "compose"), ("child", "shell")] { + app.apply(Update::ToolStarted { + id: id.into(), + title: title.into(), + kind: ToolKind::Other, + script: Some("return 1".into()), + backgrounded: false, + }); + } + app.apply(Update::ToolParent { + id: "child".into(), + parent: Some("owner".into()), + }); + for block in &mut app.blocks { + if let Block::Tool(call) = block { + call.started = std::time::Instant::now() - Duration::from_secs(100); + } + } + app.apply(Update::AgentMessage { + id: "long".into(), + text: "history\n".repeat(100), + append: false, + }); + let _ = render(&mut app, 40, 16); + assert!(app.transcript_cache[1].as_ref().unwrap().rows.is_empty()); + app.apply(Update::ToolPatched { + id: "child".into(), + title: None, + kind: None, + status: None, + script: None, + output: Some(vec!["updated while offscreen".into()]), + images: None, + append_output: true, + intent: None, + backgrounded: false, + }); + let _ = render(&mut app, 40, 16); + // Scroll to the owner through the same draw path used by the terminal. + app.follow = false; + app.scroll = 0; + let screen = render(&mut app, 40, 16); + assert!(screen.contains("Running tools"), "{screen}"); + app.apply(Update::ToolPatched { + id: "owner".into(), + title: Some("Updated owner".into()), + kind: None, + status: None, + script: None, + output: None, + images: None, + append_output: false, + intent: None, + backgrounded: false, + }); + let screen = render(&mut app, 40, 16); + assert!(screen.contains("Updated owner"), "{screen}"); + assert_transcript_rows_equal( + &app.transcript_cache[0].as_ref().unwrap().rows, + &super::transcript_block_rows(&app, 0, app.transcript_cache_width, false).0, + ); + app.follow = true; + app.apply(Update::AgentMessage { + id: "long".into(), + text: "short".into(), + append: false, + }); + let screen = render(&mut app, 40, 16); + assert!(screen.contains("short"), "{screen}"); + assert_transcript_rows_equal( + &app.transcript_cache[0].as_ref().unwrap().rows, + &super::transcript_block_rows(&app, 0, app.transcript_cache_width, false).0, + ); + assert!(app.transcript_cache[1].as_ref().unwrap().rows.is_empty()); + } + + #[test] + fn running_thought_width_and_elapsed_reflow_matches_full_layout() { + let mut app = App::new( + PathBuf::from("/tmp/kit"), + "test".into(), + "test".into(), + "0:0".into(), + ); + app.apply(Update::AgentThought { + id: "thought".into(), + text: "reasoning text".into(), + append: true, + }); + for width in [40, 18, 40] { + if let Block::Thought { started, .. } = &mut app.blocks[0] { + *started = std::time::Instant::now() - Duration::from_secs(100); + } + let _ = render(&mut app, width, 16); + assert_transcript_rows_equal( + &app.transcript_cache[0].as_ref().unwrap().rows, + &super::transcript_block_rows(&app, 0, app.transcript_cache_width, false).0, + ); + assert_eq!( + app.transcript_prefixes.last().copied(), + Some(app.transcript_cache[0].as_ref().unwrap().rows.len()) + ); + } + } + + #[test] + fn streamed_layout_matches_full_renderer_across_structural_suffixes() { + for reserve_images in [false, true] { + for suffix in [ + "more plain text\nlast line", + "# Heading\n- **bold item**\n1. Ordered item\n> quoted text\nend", + "| column | value |\n| --- | --- |\n| a | longer value |", + "```rust\nlet x = 1;\n```\nafter", + "~~~\nunfinished code\nmore", + "[linked words](https://example.com) ![plot](plot.png) end", + "![multiline\nalt](plot.png) after\nmore", + "[multiline\nlink](https://example.com) after", + "**bold** and `inline`\nnext", + ] { + let mut app = App::new( + PathBuf::from("/tmp/kit"), + "test".into(), + "test".into(), + "0:0".into(), + ); + let mut images = if reserve_images { + ImageRuntime::with_picker(Picker::halfblocks()) + } else { + ImageRuntime::disabled() + }; + let source = format!("Stable prose with unicode café.\nAnother line.\n{suffix}"); + for ch in source.chars() { + app.apply(Update::AgentMessage { + id: "stream".into(), + text: ch.to_string(), + append: true, + }); + refresh_transcript_cache_with_images(&mut app, &mut images, 19); + let (expected, placements) = + super::transcript_block_rows(&app, 0, 19, reserve_images); + let cached = app.transcript_cache[0].as_ref().unwrap(); + assert_transcript_rows_equal(&cached.rows, &expected); + assert_eq!( + cached + .images + .iter() + .map(|p| (p.row, &p.destination)) + .collect::>(), + placements + .iter() + .map(|p| (p.row, &p.destination)) + .collect::>() + ); + } + // Both resizing and replacement must discard incompatible prefixes. + for (text, width) in [(source.as_str(), 9), ("Replacement\n```\ncode", 9)] { + app.apply(Update::AgentMessage { + id: "stream".into(), + text: text.into(), + append: false, + }); + refresh_transcript_cache_with_images(&mut app, &mut images, width); + assert_transcript_rows_equal( + &app.transcript_cache[0].as_ref().unwrap().rows, + &super::transcript_block_rows(&app, 0, width, reserve_images).0, + ); + } + } + } + } + #[test] fn structured_assistant_images_share_viewports_with_markdown_images() { let parts = vec![ @@ -5052,6 +5481,56 @@ mod tests { assert!(rows[bottom].starts_with('╰')); } + #[test] + fn replacing_code_before_next_frame_invalidates_old_copy_targets() { + use crossterm::event::MouseButton; + let mut app = App::new( + PathBuf::from("/tmp/kit"), + "test".into(), + "test".into(), + "0:0".into(), + ); + app.apply(Update::AgentMessage { + id: "code".into(), + text: "```\nold code\n```".into(), + append: false, + }); + let _ = render(&mut app, 80, 24); + let row = app.row_code.iter().position(Option::is_some).unwrap(); + app.apply(Update::AgentMessage { + id: "code".into(), + text: "```\nnew code\n```".into(), + append: false, + }); + for kind in [ + MouseEventKind::Down(MouseButton::Left), + MouseEventKind::Up(MouseButton::Left), + ] { + let action = app.handle_mouse(MouseEvent { + kind, + column: app.transcript_left as u16, + row: (app.transcript_top + row) as u16, + modifiers: KeyModifiers::NONE, + }); + assert!(!matches!(action, Action::Copy(_))); + } + let _ = render(&mut app, 80, 24); + let row = app.row_code.iter().position(Option::is_some).unwrap(); + let mut action = Action::None; + for kind in [ + MouseEventKind::Down(MouseButton::Left), + MouseEventKind::Up(MouseButton::Left), + ] { + action = app.handle_mouse(MouseEvent { + kind, + column: app.transcript_left as u16, + row: (app.transcript_top + row) as u16, + modifiers: KeyModifiers::NONE, + }); + } + assert!(matches!(action, Action::Copy(text) if text == "new code")); + } + #[test] fn clicking_a_code_block_copies_exact_content() { use crossterm::event::{KeyModifiers, MouseButton, MouseEvent, MouseEventKind}; @@ -5665,7 +6144,7 @@ mod tests { refresh_transcript_cache(&mut app, 12); let history_rows = app.transcript_cache[0].as_ref().unwrap().rows.as_ptr(); let history_revision = app.transcript_cache[0].as_ref().unwrap().revision; - let tail_rows = app.transcript_cache[99].as_ref().unwrap().rows.as_ptr(); + let tail_revision = app.transcript_cache[99].as_ref().unwrap().revision; app.apply(Update::test_text(" changed".into())); refresh_transcript_cache(&mut app, 12); @@ -5674,8 +6153,8 @@ mod tests { assert_eq!(history.rows.as_ptr(), history_rows); assert_eq!(history.revision, history_revision); assert_ne!( - app.transcript_cache[99].as_ref().unwrap().rows.as_ptr(), - tail_rows + app.transcript_cache[99].as_ref().unwrap().revision, + tail_revision ); let tail = app.transcript_cache[99].as_ref().unwrap(); let text = tail From 8bc92456ac1c2d76e1f23cb3565601eb81b37207 Mon Sep 17 00:00:00 2001 From: daniel Date: Mon, 21 Sep 2026 08:55:52 -0700 Subject: [PATCH 2/3] fix(tui): prioritize input feedback over streaming work --- src/tui/mod.rs | 201 ++++++++++++--------- src/tui/scheduler.rs | 421 ++++++++++++++++++++++++++++++++++++++----- 2 files changed, 494 insertions(+), 128 deletions(-) diff --git a/src/tui/mod.rs b/src/tui/mod.rs index d457e94..5fe3162 100644 --- a/src/tui/mod.rs +++ b/src/tui/mod.rs @@ -1852,8 +1852,11 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( Update(Option), Tick, Frame, + Forward, Stop, } + let mut forward_retry = tokio::time::interval(Duration::from_millis(2)); + forward_retry.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); let mut next_priority = 0; let mut switches_closed = false; let mut pending_update = None; @@ -1868,25 +1871,7 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( transition_session.lock().map(|active| active.generation).ok(), &app.clipboard_route(), ); - if let Some(queued) = pending_update.take() { - pending_update = match scheduler::forward_updates( - &background_workers, &mut updates_rx, queued, - ) { - Ok(pending) => pending, - Err(()) => return Ok(()), - }; - } - if frames.ready(tokio::time::Instant::now()) { - draw_frame(&mut terminal, &mut app, &mut images) - .map_err(agent_client_protocol::Error::into_internal_error)?; - frames.drawn(tokio::time::Instant::now()); - } - let event = if std::mem::take(&mut submit_after_paste) { - SessionEvent::Terminal(Some(Ok(Event::Key(KeyEvent::new( - KeyCode::Enter, - KeyModifiers::NONE, - ))))) - } else { + let event = { let redraw = app.needs_redraw_tick() || images.pending(); let frame_deadline = frames.deadline(); let mut frame = pin!(tokio::time::sleep_until( @@ -1894,21 +1879,31 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( )); let mut stopped = pin!(stop.requested()); let mut shutdown = pin!(storage_shutdown.cancelled()); - // A local round-robin race keeps hot input/update queues - // from starving stop, storage, model switching or ticks. - // Return on the first Ready so no losing source consumes - // an event. Pending registers every eligible source with - // the real task waker; cancellation futures stay pinned - // until this wait ends. This scope drops all losers and - // releases input borrows before handlers drain/reset them. + // Safety first, then present the previous input before + // accepting more work. Synthetic submit remains ahead of + // real terminal events and all background completions. + // Poll crossterm with the real task waker, never a noop + // probe; return on first Ready without consuming losers. poll_fn(|cx| { - let sources = 9; + if let Poll::Ready(event) = scheduler::poll_priority( + cx, shutdown.as_mut(), stopped.as_mut(), &mut events, + &frames, &mut submit_after_paste, + ) { + return Poll::Ready(match event { + scheduler::PriorityEvent::Shutdown => SessionEvent::StorageShutdown, + scheduler::PriorityEvent::Stop => SessionEvent::Stop, + scheduler::PriorityEvent::Frame => SessionEvent::Frame, + scheduler::PriorityEvent::Terminal(event) => SessionEvent::Terminal(event), + }); + } + // Only background sources rotate. A ready paced frame + // is rendered by its branch, after the input check. + let sources = 7; for offset in 0..sources { let branch = (next_priority + offset) % sources; let ready = match branch { - 0 => shutdown.as_mut().poll(cx).map(|()| SessionEvent::StorageShutdown), - 1 => events.poll_next_unpin(cx).map(SessionEvent::Terminal), - 2 if !switches_closed => match switch_rx.poll_recv(cx) { + 0 if pending_update.is_some() => forward_retry.poll_tick(cx).map(|_| SessionEvent::Forward), + 1 if !switches_closed => match switch_rx.poll_recv(cx) { Poll::Ready(Some(completion)) => Poll::Ready(SessionEvent::ModelSwitch(completion)), Poll::Ready(None) => { // Like the former Some pattern, EOF @@ -1918,13 +1913,12 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( } Poll::Pending => Poll::Pending, }, - 3 => background_rx.poll_recv(cx).map(SessionEvent::Background), - 4 if pending_update.is_none() => updates_rx.poll_recv(cx).map(SessionEvent::Update), - 5 if redraw => ticker.poll_tick(cx).map(|_| SessionEvent::Tick), - 6 => stopped.as_mut().poll(cx).map(|()| SessionEvent::Stop), + 2 => background_rx.poll_recv(cx).map(SessionEvent::Background), + 3 if pending_update.is_none() => updates_rx.poll_recv(cx).map(SessionEvent::Update), + 4 if redraw => ticker.poll_tick(cx).map(|_| SessionEvent::Tick), - 7 => voice.poll(cx).map(SessionEvent::Voice), - 8 if frame_deadline.is_some() => frame.as_mut().poll(cx).map(|()| SessionEvent::Frame), + 5 => voice.poll(cx).map(SessionEvent::Voice), + 6 if frame_deadline.is_some() => frame.as_mut().poll(cx).map(|()| SessionEvent::Frame), _ => Poll::Pending, }; if ready.is_ready() { @@ -1937,7 +1931,9 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( }; // Invalidate before handlers: early `continue`s can also mutate // visible state. Worker forwarding alone never requests a frame. - if matches!(&event, SessionEvent::Terminal(_) | SessionEvent::Voice(_) + if matches!(&event, SessionEvent::Terminal(_)) { + frames.invalidate_input(); + } else if matches!(&event, SessionEvent::Voice(_) | SessionEvent::ModelSwitch(_) | SessionEvent::Tick) { frames.invalidate(); } @@ -2559,7 +2555,9 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( &mut app, &transition_session, &mut voice, &mut clipboard_pastes, &mut background_rx, completion, ); - if applied.dirty { + if applied.urgent { + frames.invalidate_input(); + } else if applied.dirty { frames.invalidate(); } submit_after_paste = applied.submit_after_paste; @@ -2608,7 +2606,24 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( } None => return Ok(()), }, - SessionEvent::Frame => {}, + SessionEvent::Forward => { + if let Some(queued) = pending_update.take() { + pending_update = match scheduler::forward_updates( + &background_workers, &mut updates_rx, queued, + ) { + Ok(pending) => pending, + Err(()) => return Ok(()), + }; + } + }, + SessionEvent::Frame => { + let started = tokio::time::Instant::now(); + if frames.ready(started) { + draw_frame(&mut terminal, &mut app, &mut images) + .map_err(agent_client_protocol::Error::into_internal_error)?; + frames.drawn(started); + } + }, SessionEvent::Tick => app.tick(), SessionEvent::Stop => return Ok(()), } @@ -2870,6 +2885,12 @@ fn handle_with_clipboard(app: &mut App, pastes: &mut ClipboardPastes, event: Eve handle(app, event) } +struct ClipboardPasteOutcome { + // Includes current-route errors/toasts, but never stale results. + accepted: bool, + submit: bool, +} + fn finish_clipboard_paste( app: &mut App, active: &Arc>, @@ -2877,14 +2898,17 @@ fn finish_clipboard_paste( generation: u64, route: ClipboardRoute, result: ClipboardResult, -) -> bool { - let accepted = apply_clipboard_completion(app, active, generation, route.clone(), result); - let submit = pastes.finish(generation, &route, accepted); +) -> ClipboardPasteOutcome { + let applied = apply_clipboard_completion(app, active, generation, route.clone(), result); + let submit = pastes.finish(generation, &route, applied == Some(true)); if submit { // The deferred Enter is an explicit submission, not a pasted newline. app.last_key = None; } - submit + ClipboardPasteOutcome { + accepted: applied.is_some(), + submit, + } } /// Applies one terminal event, returning the work it asks for. @@ -2980,21 +3004,23 @@ fn read_clipboard(route: &ClipboardRoute, mode: ClipboardMode) -> ClipboardResul } } +// None rejects a stale/blocked route. Some(false) reports a current-route +// failure that still needs immediate visual feedback, but must not submit. fn apply_clipboard_completion( app: &mut App, active: &Arc>, generation: u64, route: ClipboardRoute, result: ClipboardResult, -) -> bool { +) -> Option { let current_generation = active.lock().map(|route| route.generation).ok(); if current_generation != Some(generation) || app.clipboard_route() != route || paste_blocked(app) { - return false; + return None; } - match result { + Some(match result { ClipboardResult::NoImage => true, ClipboardResult::Text(text) => handle_paste(app, &text), ClipboardResult::Attachment(attachment) => attach_pasted(app, vec![attachment]), @@ -3002,7 +3028,7 @@ fn apply_clipboard_completion( app.note(error); false } - } + }) } fn handle_paste(app: &mut App, text: &str) -> bool { @@ -6168,14 +6194,10 @@ mod tests { } else { ClipboardResult::NoImage }; - assert!(super::finish_clipboard_paste( - &mut app, - &active, - &mut pastes, - 1, - route, - result - )); + assert!( + super::finish_clipboard_paste(&mut app, &active, &mut pastes, 1, route, result) + .submit + ); let Action::Submit { prompt, .. } = super::handle_with_clipboard(&mut app, &mut pastes, enter) else { @@ -6303,14 +6325,17 @@ mod tests { let mut release = KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE); release.kind = crossterm::event::KeyEventKind::Release; super::handle_with_clipboard(&mut app, &mut pastes, Event::Key(release)); - assert!(!super::finish_clipboard_paste( - &mut app, - &active, - &mut pastes, - 1, - first, - ClipboardResult::Text(" this".into()) - )); + assert!( + !super::finish_clipboard_paste( + &mut app, + &active, + &mut pastes, + 1, + first, + ClipboardResult::Text(" this".into()) + ) + .submit + ); assert_eq!(app.editor.text(), "describe this"); let attachment = clipboard_image_attachment(arboard::ImageData { width: 1, @@ -6318,14 +6343,17 @@ mod tests { bytes: Cow::Borrowed(&[20, 40, 60, 255]), }) .unwrap(); - assert!(super::finish_clipboard_paste( - &mut app, - &active, - &mut pastes, - 1, - second, - ClipboardResult::Attachment(attachment) - )); + assert!( + super::finish_clipboard_paste( + &mut app, + &active, + &mut pastes, + 1, + second, + ClipboardResult::Attachment(attachment) + ) + .submit + ); let Action::Submit { prompt, .. } = super::handle_with_clipboard(&mut app, &mut pastes, enter) else { @@ -6380,14 +6408,10 @@ mod tests { } else { ClipboardResult::Error("clipboard unavailable".into()) }; - assert!(!super::finish_clipboard_paste( - &mut app, - &active, - &mut pastes, - 1, - route, - result - )); + assert!( + !super::finish_clipboard_paste(&mut app, &active, &mut pastes, 1, route, result) + .submit + ); assert_eq!(app.editor.text(), before); assert!(pastes.pending.is_none()); } @@ -6420,14 +6444,17 @@ mod tests { } else { super::handle_with_clipboard(&mut app, &mut pastes, Event::Paste(" edited".into())); } - assert!(!super::finish_clipboard_paste( - &mut app, - &active, - &mut pastes, - 1, - route, - ClipboardResult::Text(" pasted".into()) - )); + assert!( + !super::finish_clipboard_paste( + &mut app, + &active, + &mut pastes, + 1, + route, + ClipboardResult::Text(" pasted".into()) + ) + .submit + ); assert_eq!( app.editor.text(), if switch { diff --git a/src/tui/scheduler.rs b/src/tui/scheduler.rs index 80257ee..dda21a5 100644 --- a/src/tui/scheduler.rs +++ b/src/tui/scheduler.rs @@ -8,11 +8,39 @@ use std::sync::{Arc, Mutex}; use tokio::{sync::mpsc, time::Instant}; const FRAME_INTERVAL: std::time::Duration = std::time::Duration::from_millis(16); -// Yield to the round-robin event selector even when producers keep queues hot. +// Yield to input and safety checks even when producers keep queues hot. const UPDATE_BURST: usize = 64; +const UPDATE_BUDGET: std::time::Duration = std::time::Duration::from_millis(2); + +/// Checked between records, before consuming the next queue entry. A single +/// synchronous record can exceed this budget; it is not a latency guarantee. +struct BatchBudget { + started: Instant, + records: usize, +} + +impl BatchBudget { + fn new(started: Instant) -> Self { + Self { + started, + records: 0, + } + } + + fn take(&mut self, now: Instant) -> bool { + if self.records >= UPDATE_BURST + || (self.records > 0 && now.duration_since(self.started) >= UPDATE_BUDGET) + { + return false; + } + self.records += 1; + true + } +} pub(super) struct Frames { dirty: bool, + urgent: bool, next: Instant, } @@ -20,6 +48,7 @@ impl Frames { pub(super) fn new(now: Instant) -> Self { Self { dirty: true, + urgent: false, next: now, } } @@ -28,21 +57,69 @@ impl Frames { self.dirty = true; } + pub(super) fn invalidate_input(&mut self) { + self.dirty = true; + self.urgent = true; + } + + pub(super) fn urgent(&self) -> bool { + self.urgent + } + pub(super) fn deadline(&self) -> Option { self.dirty.then_some(self.next) } pub(super) fn ready(&self, now: Instant) -> bool { - self.dirty && now >= self.next + self.dirty && (self.urgent || now >= self.next) } pub(super) fn drawn(&mut self, now: Instant) { self.dirty = false; - // Do not catch up missed frames after slow draws or blocking actions. + self.urgent = false; + // Pace from the start of the draw, without catching up missed frames. self.next = now + FRAME_INTERVAL; } } +/// The priority tier polls real event sources with the caller's task waker. +/// A pending result permits the caller to poll the background round-robin tier. +pub(super) enum PriorityEvent { + Shutdown, + Stop, + Frame, + Terminal(Option>), +} + +pub(super) fn poll_priority( + cx: &mut std::task::Context<'_>, + mut shutdown: std::pin::Pin<&mut impl std::future::Future>, + mut stopped: std::pin::Pin<&mut impl std::future::Future>, + events: &mut (impl futures_util::Stream> + Unpin), + frames: &Frames, + submit_after_paste: &mut bool, +) -> std::task::Poll { + use crossterm::event::{Event, KeyCode, KeyEvent, KeyModifiers}; + use futures_util::StreamExt; + use std::task::Poll; + + if shutdown.as_mut().poll(cx).is_ready() { + return Poll::Ready(PriorityEvent::Shutdown); + } + if stopped.as_mut().poll(cx).is_ready() { + return Poll::Ready(PriorityEvent::Stop); + } + if frames.urgent() { + return Poll::Ready(PriorityEvent::Frame); + } + if std::mem::take(submit_after_paste) { + return Poll::Ready(PriorityEvent::Terminal(Some(Ok(Event::Key( + KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE), + ))))); + } + events.poll_next_unpin(cx).map(PriorityEvent::Terminal) +} + /// Keep a full worker queue's head ahead of all later updates. No update bypasses /// image materialization, including non-image updates following an image. pub(super) fn forward_updates( @@ -50,10 +127,12 @@ pub(super) fn forward_updates( updates: &mut mpsc::UnboundedReceiver, first: QueuedUpdate, ) -> Result, ()> { - for queued in std::iter::once(first) - .chain(std::iter::from_fn(|| updates.try_recv().ok())) - .take(UPDATE_BURST) - { + let mut budget = BatchBudget::new(Instant::now()); + let mut first = Some(first); + while budget.take(Instant::now()) { + let Some(queued) = first.take().or_else(|| updates.try_recv().ok()) else { + break; + }; match workers.try_update(queued) { Ok(()) => {} Err(error) => match *error { @@ -67,6 +146,7 @@ pub(super) fn forward_updates( pub(super) struct Applied { pub(super) dirty: bool, + pub(super) urgent: bool, pub(super) submit_after_paste: bool, } @@ -82,12 +162,15 @@ pub(super) fn apply_completions( ) -> Applied { let mut applied = Applied { dirty: false, + urgent: false, submit_after_paste: false, }; - for completion in std::iter::once(first) - .chain(std::iter::from_fn(|| completed.try_recv().ok())) - .take(UPDATE_BURST) - { + let mut budget = BatchBudget::new(Instant::now()); + let mut first = Some(first); + while budget.take(Instant::now()) { + let Some(completion) = first.take().or_else(|| completed.try_recv().ok()) else { + break; + }; match completion { BackgroundCompletion::Update { queued, images } => { if let Some(update) = accept_queued_update(route, queued) { @@ -104,12 +187,15 @@ pub(super) fn apply_completions( route: clipboard_route, result, } => { - applied.submit_after_paste = + let outcome = finish_clipboard_paste(app, route, pastes, generation, clipboard_route, result); - applied.dirty = true; - // The synthetic Enter must precede the next completion, just as - // it did when the active loop consumed one completion per turn. - if applied.submit_after_paste { + if outcome.accepted { + applied.submit_after_paste = outcome.submit; + applied.dirty = true; + applied.urgent = true; + // Present paste feedback before consuming more streaming work, + // even without deferred Enter. A synthetic submit still precedes + // the next completion after this urgent frame. break; } } @@ -146,7 +232,7 @@ mod tests { assert_eq!(frames.deadline(), Some(deadline)); assert!(!frames.ready(now)); assert!(frames.ready(deadline)); - // A late frame schedules from its completion, not from the old deadline. + // A late frame schedules from its start, not from the old deadline. let late = deadline + FRAME_INTERVAL; frames.drawn(late); frames.invalidate(); @@ -154,6 +240,166 @@ mod tests { assert!(frames.ready(late + FRAME_INTERVAL)); } + #[test] + fn input_frame_bypasses_pacing_and_consumes_pending_stream_invalidation() { + let now = Instant::now(); + let mut frames = Frames::new(now); + frames.drawn(now); + frames.invalidate(); + let input_at = now + std::time::Duration::from_millis(1); + assert!(!frames.ready(input_at)); + frames.invalidate_input(); + assert!(frames.urgent()); + assert!(frames.ready(input_at)); + frames.drawn(input_at); + assert!(!frames.urgent()); + assert_eq!(frames.deadline(), None); + assert!(!frames.ready(now + FRAME_INTERVAL)); + frames.invalidate(); + assert_eq!(frames.deadline(), Some(input_at + FRAME_INTERVAL)); + assert!(!frames.ready(input_at)); + } + + #[test] + fn batch_budget_checks_elapsed_time_and_count_between_records() { + let now = Instant::now(); + let mut budget = BatchBudget::new(now); + assert!(budget.take(now)); + assert!(budget.take(now + UPDATE_BUDGET / 2)); + assert!(!budget.take(now + UPDATE_BUDGET)); + assert!(!budget.take(now + UPDATE_BUDGET * 2)); + let mut budget = BatchBudget::new(now); + // The cap is an explicit policy, not an observed production work count. + for _ in 0..UPDATE_BURST { + assert!(budget.take(now)); + } + assert!(!budget.take(now)); + } + + #[tokio::test] + async fn priority_tier_preserves_safety_input_and_synthetic_submit_order() { + use crossterm::event::{Event, KeyCode, KeyEvent, KeyModifiers}; + use std::{ + future::{pending, poll_fn, ready}, + pin::pin, + }; + let (sender, mut receiver) = mpsc::unbounded_channel(); + let key = Event::Key(KeyEvent::new(KeyCode::Char('x'), KeyModifiers::NONE)); + sender.send(Ok(key.clone())).unwrap(); + let mut events = futures_util::stream::poll_fn(|cx| receiver.poll_recv(cx)); + let mut frames = Frames::new(Instant::now()); + frames.invalidate_input(); + let mut submit = true; + let mut shutdown = pin!(ready(())); + let mut stopped = pin!(ready(())); + assert!(matches!( + poll_fn(|cx| poll_priority( + cx, + shutdown.as_mut(), + stopped.as_mut(), + &mut events, + &frames, + &mut submit + )) + .await, + PriorityEvent::Shutdown + )); + let mut shutdown = pin!(pending()); + assert!(matches!( + poll_fn(|cx| poll_priority( + cx, + shutdown.as_mut(), + stopped.as_mut(), + &mut events, + &frames, + &mut submit + )) + .await, + PriorityEvent::Stop + )); + let mut stopped = pin!(pending()); + assert!(matches!( + poll_fn(|cx| poll_priority( + cx, + shutdown.as_mut(), + stopped.as_mut(), + &mut events, + &frames, + &mut submit + )) + .await, + PriorityEvent::Frame + )); + assert!(submit); + frames.drawn(Instant::now() - FRAME_INTERVAL); + frames.invalidate(); // Even a due stream frame must wait for input. + assert!(frames.ready(Instant::now())); + assert!(matches!( + poll_fn(|cx| poll_priority( + cx, + shutdown.as_mut(), + stopped.as_mut(), + &mut events, + &frames, + &mut submit + )) + .await, + PriorityEvent::Terminal(Some(Ok(Event::Key(KeyEvent { + code: KeyCode::Enter, + .. + })))) + )); + assert!(!submit); + let event = poll_fn(|cx| { + poll_priority( + cx, + shutdown.as_mut(), + stopped.as_mut(), + &mut events, + &frames, + &mut submit, + ) + }) + .await; + assert!(matches!(event, PriorityEvent::Terminal(Some(Ok(event))) if event == key)); + // With no input, the production selector may now service background work. + poll_fn(|cx| { + assert!( + poll_priority( + cx, + shutdown.as_mut(), + stopped.as_mut(), + &mut events, + &frames, + &mut submit + ) + .is_pending() + ); + std::task::Poll::Ready(()) + }) + .await; + // The same source must also wake a pending selector, not just win when + // already queued. No synthetic waker or polling-count assertion. + let receive = poll_fn(|cx| { + poll_priority( + cx, + shutdown.as_mut(), + stopped.as_mut(), + &mut events, + &frames, + &mut submit, + ) + }); + let send = async { + tokio::task::yield_now().await; + sender.send(Ok(Event::Paste("new paste".into()))).unwrap(); + }; + let (event, ()) = tokio::join!(receive, send); + assert!( + matches!(event, PriorityEvent::Terminal(Some(Ok(Event::Paste(text)))) if text == "new paste") + ); + } + #[test] fn forwarding_preserves_order_across_worker_backpressure() { let (sender, worker) = std::sync::mpsc::sync_channel(1); @@ -179,7 +425,9 @@ mod tests { panic!("expected log") }; received.push(text); - let Some(first) = pending else { break }; + let Some(first) = pending.or_else(|| updates.try_recv().ok()) else { + break; + }; pending = forward_updates(&workers, &mut updates, first).unwrap(); } assert_eq!(received, ["one", "two", "three"]); @@ -220,11 +468,15 @@ mod tests { let next = updates .try_recv() .expect("a bounded turn leaves queued work"); - assert!( - forward_updates(&workers, &mut updates, next) - .unwrap() - .is_none() - ); + let mut next = Some(next); + while let Some(first) = next { + assert!( + forward_updates(&workers, &mut updates, first) + .unwrap() + .is_none() + ); + next = updates.try_recv().ok(); + } let received: Vec<_> = worker .try_iter() .map(|queued| { @@ -250,6 +502,24 @@ mod tests { let mut pastes = ClipboardPastes::default(); pastes.queued(7, clipboard_route.clone()); let (sender, mut completed) = mpsc::channel(2); + let applied = apply_completions( + &mut app, + &route(), + &mut NativeVoice::default(), + &mut pastes, + &mut completed, + BackgroundCompletion::Clipboard { + generation: 6, + route: clipboard_route.clone(), + result: ClipboardResult::Text("stale".into()), + }, + ); + assert!(!applied.dirty); + assert!(!applied.submit_after_paste); + assert!(!applied.urgent); + assert_eq!(app.editor.text(), ""); + assert!(pastes.pending.is_some()); + // Stale results cannot finish the current paste or request a frame. sender .try_send(BackgroundCompletion::Clipboard { generation: 7, @@ -258,19 +528,16 @@ mod tests { }) .ok() .unwrap(); + let first = completed.try_recv().unwrap(); let applied = apply_completions( &mut app, &route(), &mut NativeVoice::default(), &mut pastes, &mut completed, - BackgroundCompletion::Clipboard { - generation: 6, - route: clipboard_route, - result: ClipboardResult::Text("stale".into()), - }, + first, ); - assert!(applied.dirty); + assert!(applied.urgent); assert!(!applied.submit_after_paste); assert_eq!(app.editor.text(), "current"); assert!(pastes.pending.is_none()); @@ -353,18 +620,23 @@ mod tests { )) .ok() .unwrap(); - let first = completed.try_recv().unwrap(); - let applied = apply_completions( - &mut app, - &route(), - &mut voice, - &mut ClipboardPastes::default(), - &mut completed, - first, - ); - assert!(applied.dirty); - assert!(!applied.submit_after_paste); - assert!(completed.try_recv().is_err()); + let route = route(); + let mut pastes = ClipboardPastes::default(); + let mut dirty = false; + while let Ok(first) = completed.try_recv() { + let applied = apply_completions( + &mut app, + &route, + &mut voice, + &mut pastes, + &mut completed, + first, + ); + dirty |= applied.dirty; + assert!(!applied.urgent); + assert!(!applied.submit_after_paste); + } + assert!(dirty); let handoff = voice.handoff.as_ref().unwrap(); assert!(handoff.accepted); assert_eq!(handoff.user_message_id.as_deref(), Some("user")); @@ -388,6 +660,72 @@ mod tests { assert!(app.blocks.is_empty()); } + #[test] + fn clipboard_without_submit_requests_urgent_frame_and_leaves_stream_queued() { + let mut app = app(); + let clipboard_route = app.clipboard_route(); + let mut pastes = ClipboardPastes::default(); + pastes.queued(7, clipboard_route.clone()); + let (sender, mut completed) = mpsc::channel(2); + sender + .try_send(completion(7, Update::Log("after paste".into()))) + .ok() + .unwrap(); + let applied = apply_completions( + &mut app, + &route(), + &mut NativeVoice::default(), + &mut pastes, + &mut completed, + BackgroundCompletion::Clipboard { + generation: 7, + route: clipboard_route, + result: ClipboardResult::Text("pasted".into()), + }, + ); + assert!(applied.dirty); + assert!(applied.urgent); + assert!(!applied.submit_after_paste); + assert_eq!(app.editor.text(), "pasted"); + assert!(pastes.pending.is_none()); + assert!(app.logs.is_empty()); + let BackgroundCompletion::Update { queued, .. } = completed.try_recv().unwrap() else { + panic!("expected queued streaming update"); + }; + assert!(matches!(queued.update, Update::Log(text) if text == "after paste")); + } + + #[test] + fn current_clipboard_error_is_urgent_but_cannot_submit() { + let mut app = app(); + let clipboard_route = app.clipboard_route(); + let mut pastes = ClipboardPastes::default(); + pastes.queued(7, clipboard_route.clone()); + pastes.pending.as_mut().unwrap().submit = true; + let (sender, mut completed) = mpsc::channel(2); + sender + .try_send(completion(7, Update::Log("after paste".into()))) + .ok() + .unwrap(); + let applied = apply_completions( + &mut app, + &route(), + &mut NativeVoice::default(), + &mut pastes, + &mut completed, + BackgroundCompletion::Clipboard { + generation: 7, + route: clipboard_route, + result: ClipboardResult::Error("clipboard unavailable".into()), + }, + ); + assert!(applied.urgent); + assert!(applied.dirty); + assert!(!applied.submit_after_paste); + assert!(pastes.pending.is_none()); + assert!(completed.try_recv().is_ok()); + } + #[test] fn clipboard_submission_is_a_batch_boundary() { let mut app = app(); @@ -414,6 +752,7 @@ mod tests { ); assert!(applied.dirty); assert!(applied.submit_after_paste); + assert!(applied.urgent); assert_eq!(app.editor.text(), "pasted"); assert!(completed.try_recv().is_ok()); assert!(app.blocks.is_empty()); From bfaa11690dc0e74fccd3347f04c00b1c46dfb760 Mon Sep 17 00:00:00 2001 From: daniel Date: Mon, 21 Sep 2026 15:12:27 -0700 Subject: [PATCH 3/3] fix(tui): scope terminal input ownership across lifecycle handoffs --- src/tui/image.rs | 12 +- src/tui/image_query.rs | 234 +++++++++++++ src/tui/image_query_tests.rs | 267 +++++++++++++++ src/tui/input.rs | 95 ++++++ src/tui/input_tests.rs | 331 +++++++++++++++++++ src/tui/mod.rs | 134 +++++--- src/tui/startup.rs | 6 +- tests/support/terminal_latency/README.md | 72 ++++ tests/support/terminal_latency/driver.rs | 26 ++ tests/support/terminal_latency/probe.py | 195 +++++++++++ tests/support/terminal_latency/test_probe.py | 45 +++ 11 files changed, 1370 insertions(+), 47 deletions(-) create mode 100644 src/tui/image_query.rs create mode 100644 src/tui/image_query_tests.rs create mode 100644 src/tui/input.rs create mode 100644 src/tui/input_tests.rs create mode 100644 tests/support/terminal_latency/README.md create mode 100644 tests/support/terminal_latency/driver.rs create mode 100644 tests/support/terminal_latency/probe.py create mode 100644 tests/support/terminal_latency/test_probe.py diff --git a/src/tui/image.rs b/src/tui/image.rs index 8b49446..42fe2ec 100644 --- a/src/tui/image.rs +++ b/src/tui/image.rs @@ -14,13 +14,14 @@ use ratatui::{ }; use ratatui_image::{ Resize, - picker::{Picker, ProtocolType, cap_parser::QueryStdioOptions}, + picker::Picker, sliced::{SignedPosition, SlicedImage, SlicedProtocol}, }; use super::app::UserImage; -const TERMINAL_QUERY_TIMEOUT: Duration = Duration::from_millis(150); +#[path = "image_query.rs"] +mod image_query; const MAX_DECODED_ALLOCATION: u64 = 64 * 1024 * 1024; const MAX_DECODED_BACKING_BYTES: u64 = 128 * 1024 * 1024; const MAX_CACHE_ENTRIES: usize = 16; @@ -49,12 +50,7 @@ pub(super) struct ImageRuntime { impl ImageRuntime { pub fn detect() -> Self { - let picker = Picker::from_query_stdio_with_options(QueryStdioOptions { - timeout: TERMINAL_QUERY_TIMEOUT, - ..QueryStdioOptions::default() - }) - .ok() - .filter(|picker| picker.protocol_type() != ProtocolType::Halfblocks); + let picker = image_query::detect(); Self { picker, cache: HashMap::new(), diff --git a/src/tui/image_query.rs b/src/tui/image_query.rs new file mode 100644 index 0000000..4c13741 --- /dev/null +++ b/src/tui/image_query.rs @@ -0,0 +1,234 @@ +//! Synchronous probing before the event reader starts. Picker's query helpers +//! detach a stdin reader on timeout and can later restore stale terminal modes. +use ratatui_image::{ + FontSize, + picker::{Picker, ProtocolType}, +}; +#[cfg(unix)] +use std::time::Duration; + +pub(super) fn detect() -> Option { + // The caller owns raw mode throughout this synchronous input interval. + // Construct before probing so tmux passthrough is enabled on first entry. + // The temporary default is for setup only, never evidence of a known font. + let fallback = terminal_font_size(); + let initial_size = fallback.unwrap_or(FontSize::new(10, 20)); + let mut picker = picker_with_font_size(initial_size); + #[allow(unused_mut)] // Only Unix collects query responses. + let mut detected = Detection::default(); + #[cfg(unix)] + { + use std::{fs::OpenOptions, os::unix::fs::OpenOptionsExt}; + // Separate open file description: never change stdin's file flags. + if let Ok(terminal) = OpenOptions::new() + .read(true) + .write(true) + .custom_flags(libc::O_NONBLOCK | libc::O_NOCTTY | libc::O_CLOEXEC) + .open("/dev/tty") + { + let tmux = std::env::var("TERM").is_ok_and(|v| v.starts_with("tmux")) + || std::env::var("TERM_PROGRAM").is_ok_and(|v| v == "tmux"); + let blacklist = ["WEZTERM_EXECUTABLE", "KONSOLE_VERSION"] + .iter() + .any(|key| std::env::var(key).is_ok_and(|v| !v.is_empty())); + detected = query(&terminal, tmux, blacklist, Duration::from_millis(150)); + } + } + // Windows: do not issue queries without a cancellable byte-input API. + // ConPTY does not reliably return replies. Preserve native iTerm2 environment + // hints when window-size metadata supplies a font size, without mode changes. + // As with the original picker fallback, no known font means no native images. + let size = detected.font_size.or(fallback)?; + if (size.width, size.height) != (initial_size.width, initial_size.height) { + picker = picker_with_font_size(size); + } + if let Some(protocol) = detected.protocol { + picker.set_protocol_type(protocol); + } + (picker.protocol_type() != ProtocolType::Halfblocks).then_some(picker) +} + +#[allow(deprecated)] // Query constructors spawn an unjoinable stdin reader. +fn picker_with_font_size(size: FontSize) -> Picker { + // This dependency constructor may synchronously wait for `tmux set -p + // allow-passthrough on`. That preexisting subprocess is outside the query IO + // deadline; public APIs cannot bypass it while preserving tmux state. It runs + // before probing and again only if the reported font differs. It never reads + // stdin or changes terminal modes. Fully bounding this setup needs an upstream + // side-effect-free constructor; do not bypass it by mutating process env. + Picker::from_fontsize(size) +} + +fn terminal_font_size() -> Option { + let size = crossterm::terminal::window_size().ok()?; + let width = size.width.checked_div(size.columns)?; + let height = size.height.checked_div(size.rows)?; + (width > 0 && height > 0).then_some(FontSize::new(width, height)) +} + +#[cfg(unix)] +const MAX_QUERY_BYTES: usize = 4096; +#[derive(Default)] +struct Detection { + font_size: Option, + protocol: Option, +} + +#[cfg(unix)] +fn query(terminal: &std::fs::File, tmux: bool, blacklist: bool, timeout: Duration) -> Detection { + use ratatui_image::picker::cap_parser::{Parser, QueryStdioOptions, Response}; + use std::{ + io::{Read, Write}, + os::fd::AsRawFd, + time::Instant, + }; + let deadline = Instant::now() + timeout; + let fd = terminal.as_raw_fd(); + let mut terminal = terminal; + let mut detected = Detection::default(); + let query = Parser::query( + tmux, + QueryStdioOptions { + timeout, + blacklist_protocols: if blacklist { + vec![ProtocolType::Kitty, ProtocolType::Sixel] + } else { + vec![] + }, + ..QueryStdioOptions::default() + }, + ); + let mut output = query.as_bytes(); + while !output.is_empty() && ready(fd, libc::POLLOUT, deadline) { + match terminal.write(output) { + Ok(0) => return detected, + Ok(n) => output = &output[n..], + Err(e) + if matches!( + e.kind(), + std::io::ErrorKind::Interrupted | std::io::ErrorKind::WouldBlock + ) => {} + Err(_) => return detected, + } + } + if !output.is_empty() { + return detected; + } + let mut parser = Parser::new(); + // Bound parser allocation/work even on malformed, unterminated replies or + // perpetually readable input. The absolute monotonic deadline never resets. + // One-byte reads stop exactly at DSR, leaving subsequent input untouched. + for _ in 0..MAX_QUERY_BYTES { + if !ready(fd, libc::POLLIN, deadline) { + break; + } + let mut byte = [0]; + match terminal.read(&mut byte) { + Ok(0) => break, + Ok(_) => {} + Err(e) + if matches!( + e.kind(), + std::io::ErrorKind::Interrupted | std::io::ErrorKind::WouldBlock + ) => + { + continue; + } + Err(_) => break, + } + for response in parser.push(char::from(byte[0])) { + match response { + Response::Status => return detected, + Response::Kitty if !blacklist => detected.protocol = Some(ProtocolType::Kitty), + Response::Sixel if !blacklist && detected.protocol.is_none() => { + detected.protocol = Some(ProtocolType::Sixel) + } + Response::CellSize(Some((w, h))) => detected.font_size = Some(FontSize::new(w, h)), + _ => {} + } + } + } + detected +} + +// Darwin's /dev/tty alias returns POLLNVAL from poll even though the same +// controlling terminal supports select. Tests on an openpty slave alone do not +// exercise that alias. Keep poll elsewhere, including its high-fd support. +#[cfg(target_os = "macos")] +fn ready(fd: std::os::fd::RawFd, events: libc::c_short, deadline: std::time::Instant) -> bool { + if fd < 0 || fd as usize >= libc::FD_SETSIZE { + return false; + } + loop { + let Some(remaining) = deadline.checked_duration_since(std::time::Instant::now()) else { + return false; + }; + if remaining.is_zero() { + return false; + } + // SAFETY: zero is a valid empty fd_set; fd was range-checked before + // FD_SET. select receives only live initialized sets and a timeval. + let result = unsafe { + let mut reads: libc::fd_set = std::mem::zeroed(); + let mut writes: libc::fd_set = std::mem::zeroed(); + if events & libc::POLLIN != 0 { + libc::FD_SET(fd, &mut reads); + } + if events & libc::POLLOUT != 0 { + libc::FD_SET(fd, &mut writes); + } + let mut timeout = libc::timeval { + tv_sec: remaining.as_secs().min(libc::time_t::MAX as u64) as libc::time_t, + tv_usec: remaining.subsec_micros() as libc::suseconds_t, + }; + libc::select( + fd + 1, + &mut reads, + &mut writes, + std::ptr::null_mut(), + &mut timeout, + ) + }; + if result >= 0 { + return result > 0 && std::time::Instant::now() < deadline; + } + if std::io::Error::last_os_error().kind() != std::io::ErrorKind::Interrupted { + return false; + } + } +} + +#[cfg(all(unix, not(target_os = "macos")))] +fn ready(fd: std::os::fd::RawFd, events: libc::c_short, deadline: std::time::Instant) -> bool { + loop { + let Some(remaining) = deadline.checked_duration_since(std::time::Instant::now()) else { + return false; + }; + if remaining.is_zero() { + return false; + } + let mut poll = libc::pollfd { + fd, + events, + revents: 0, + }; + let millis = remaining + .as_millis() + .saturating_add(1) + .min(i32::MAX as u128) as i32; + // SAFETY: poll points at one initialized, live pollfd. + let result = unsafe { libc::poll(&mut poll, 1, millis) }; + if result >= 0 { + return result > 0 + && poll.revents & events != 0 + && std::time::Instant::now() < deadline; + } + if std::io::Error::last_os_error().kind() != std::io::ErrorKind::Interrupted { + return false; + } + } +} + +#[cfg(all(test, unix))] +#[path = "image_query_tests.rs"] +mod tests; diff --git a/src/tui/image_query_tests.rs b/src/tui/image_query_tests.rs new file mode 100644 index 0000000..876bc5b --- /dev/null +++ b/src/tui/image_query_tests.rs @@ -0,0 +1,267 @@ +//! Real OS IO boundaries; assertion conveniences are confined to tests. +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::disallowed_methods, + clippy::disallowed_macros +)] + +use super::*; +use std::{ + fs::File, + io::{Read, Write}, + os::fd::{AsRawFd, FromRawFd}, +}; + +fn terminal_pair() -> (File, File) { + let (mut master, mut slave) = (-1, -1); + // SAFETY: valid output pointers; optional name/termios/winsize are null. + assert_eq!( + unsafe { + libc::openpty( + &mut master, + &mut slave, + std::ptr::null_mut(), + std::ptr::null_mut(), + std::ptr::null_mut(), + ) + }, + 0 + ); + // SAFETY: successful openpty returned uniquely owned descriptors. + let pair = unsafe { (File::from_raw_fd(master), File::from_raw_fd(slave)) }; + // SAFETY: termios is initialized by successful tcgetattr, then applied to + // this test's slave only. Nonblocking mode is required by query's IO contract. + unsafe { + let mut mode = std::mem::zeroed(); + assert_eq!(libc::tcgetattr(slave, &mut mode), 0); + libc::cfmakeraw(&mut mode); + assert_eq!(libc::tcsetattr(slave, libc::TCSANOW, &mode), 0); + assert_eq!(libc::fcntl(slave, libc::F_SETFL, libc::O_NONBLOCK), 0); + } + pair +} + +#[test] +fn native_replies_preserve_following_input() { + let (mut master, mut slave) = terminal_pair(); + master + .write_all(b"\x1b[?64;4c\x1b_Gi=31;OK\x1b\\\x1b[6;20;10t\x1b[0nX") + .unwrap(); + let result = query(&slave, false, false, Duration::from_millis(150)); + assert_eq!(result.protocol, Some(ProtocolType::Kitty)); + assert_eq!( + result.font_size.map(|s| (s.width, s.height)), + Some((10, 20)) + ); + let mut next = [0]; + assert_eq!(slave.read(&mut next).unwrap(), 1); + assert_eq!(next, *b"X"); +} + +#[test] +fn sixel_and_blacklisted_protocols() { + for blacklist in [false, true] { + let (mut master, slave) = terminal_pair(); + master.write_all(b"\x1b[?64;4c\x1b[6;16;8t\x1b[0n").unwrap(); + let result = query(&slave, false, blacklist, Duration::from_millis(150)); + assert_eq!( + result.protocol, + if blacklist { + None + } else { + Some(ProtocolType::Sixel) + } + ); + assert_eq!(result.font_size.map(|s| (s.width, s.height)), Some((8, 16))); + } +} + +#[test] +fn unanswered_and_partial_queries_release_input_without_mode_changes() { + for reply in [b"".as_slice(), b"\x1b[6;20;"] { + let (mut master, mut slave) = terminal_pair(); + master.write_all(reply).unwrap(); + let result = query(&slave, false, false, Duration::from_millis(5)); + assert_eq!(result.protocol, None); + assert!(result.font_size.is_none()); + // No detached reader remains to consume subsequent keystrokes. + master.write_all(b"X").unwrap(); + let mut next = [0]; + assert!(ready( + slave.as_raw_fd(), + libc::POLLIN, + std::time::Instant::now() + Duration::from_secs(1) + )); + assert_eq!(slave.read(&mut next).unwrap(), 1); + assert_eq!(next, *b"X"); + // SAFETY: a valid live tty and initialized output storage. + unsafe { + let mut mode = std::mem::zeroed(); + assert_eq!(libc::tcgetattr(slave.as_raw_fd(), &mut mode), 0); + assert_eq!(mode.c_lflag & (libc::ICANON | libc::ECHO), 0); + } + } +} + +#[test] +fn unsupported_status_leaves_images_unsupported() { + let (mut master, slave) = terminal_pair(); + master.write_all(b"\x1b[?1;2c\x1b[0n").unwrap(); + let result = query(&slave, false, false, Duration::from_millis(150)); + assert_eq!(result.protocol, None); +} + +#[test] +fn expired_deadline_does_not_consume_ready_input() { + let (mut master, mut slave) = terminal_pair(); + master.write_all(b"X").unwrap(); + let _ = query(&slave, false, false, Duration::ZERO); + let mut next = [0]; + assert_eq!(slave.read(&mut next).unwrap(), 1); + assert_eq!(next, *b"X"); +} + +#[test] +fn tmux_query_uses_passthrough_wrapping() { + let (mut master, slave) = terminal_pair(); + master.write_all(b"\x1b[6;16;8t\x1b[0n").unwrap(); + let result = query(&slave, true, false, Duration::from_millis(150)); + assert_eq!(result.font_size.map(|s| (s.width, s.height)), Some((8, 16))); + let mut bytes = [0; 512]; + let count = master.read(&mut bytes).unwrap(); + let output = &bytes[..count]; + assert!(output.starts_with(b"\x1bPtmux;")); + assert!(output.ends_with(b"\x1b\\")); + assert!(output.windows(6).any(|part| part == b"\x1b\x1b[16t")); +} + +#[test] +fn zero_font_dimensions_are_not_accepted() { + for response in [b"\x1b[6;0;10t\x1b[0n", b"\x1b[6;20;0t\x1b[0n"] { + let (mut master, slave) = terminal_pair(); + master.write_all(response).unwrap(); + let result = query(&slave, false, false, Duration::from_millis(150)); + assert!(result.font_size.is_none()); + } +} + +#[test] +fn malformed_input_exhausts_budget_without_consuming_trailing_reply() { + // A real nonblocking OS byte stream avoids PTY queue-size differences when + // preloading more than the query budget. No parser or IO replacement hooks. + use std::os::{fd::OwnedFd, unix::net::UnixStream}; + let (mut peer, stream) = UnixStream::pair().unwrap(); + stream.set_nonblocking(true).unwrap(); + peer.set_nonblocking(true).unwrap(); + let mut terminal = File::from(OwnedFd::from(stream)); + let mut malformed = vec![b'9'; MAX_QUERY_BYTES + 128]; + malformed[..2].copy_from_slice(b"\x1b["); + malformed.extend_from_slice(b"\x1b_Gi=31;OK\x1b\\\x1b[0n"); + peer.write_all(&malformed).unwrap(); + let result = query(&terminal, false, false, Duration::from_secs(1)); + assert!(result.protocol.is_none()); + assert!(result.font_size.is_none()); + let mut unread = Vec::new(); + let error = terminal.read_to_end(&mut unread).unwrap_err(); + assert_eq!(error.kind(), std::io::ErrorKind::WouldBlock); + assert!(unread.ends_with(b"\x1b_Gi=31;OK\x1b\\\x1b[0n")); +} + +#[test] +fn actual_detect_queries_controlling_terminal_and_enables_images() { + use std::{ + os::unix::process::CommandExt, + process::{Child, Command, Stdio}, + time::Instant, + }; + const CHILD: &str = "KIT_IMAGE_DETECT_PTY_CHILD"; + const TEST: &str = "tui::image::image_query::tests::actual_detect_queries_controlling_terminal_and_enables_images"; + if std::env::var_os(CHILD).is_some() { + crossterm::terminal::enable_raw_mode().unwrap(); + let result = detect(); + crossterm::terminal::disable_raw_mode().unwrap(); + let picker = result.expect("actual detect must negotiate native images"); + assert_eq!(picker.protocol_type(), ProtocolType::Kitty); + assert_eq!( + (picker.font_size().width, picker.font_size().height), + (8, 16) + ); + return; + } + struct ChildGuard(Child); + impl Drop for ChildGuard { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } + } + let (mut master, slave) = terminal_pair(); + let mut command = Command::new(std::env::current_exe().unwrap()); + command + .args(["--exact", TEST, "--nocapture"]) + .env(CHILD, "1") + .env("TERM", "xterm-256color") + .env_remove("TERM_PROGRAM") + .env_remove("TMUX") + .env_remove("WEZTERM_EXECUTABLE") + .env_remove("KONSOLE_VERSION") + .stdin(Stdio::from(slave.try_clone().unwrap())) + .stdout(Stdio::from(slave.try_clone().unwrap())) + .stderr(Stdio::from(slave)); + // SAFETY: only async-signal-safe calls in the fork/exec interval. The child + // opens /dev/tty through production detect(), not the parent's slave file. + unsafe { + command.pre_exec(|| { + if libc::setsid() == -1 || libc::ioctl(0, libc::TIOCSCTTY as _, 0) == -1 { + return Err(std::io::Error::last_os_error()); + } + Ok(()) + }); + } + let mut child = ChildGuard(command.spawn().unwrap()); + drop(command); + let deadline = Instant::now() + Duration::from_secs(10); + let mut output = String::new(); + let mut answered = false; + loop { + assert!( + Instant::now() < deadline, + "actual detect timed out: {output:?}" + ); + let mut fd = libc::pollfd { + fd: master.as_raw_fd(), + events: libc::POLLIN, + revents: 0, + }; + // SAFETY: poll receives one initialized pollfd for our live PTY master. + if unsafe { libc::poll(&mut fd, 1, 20) } > 0 { + let mut bytes = [0; 4096]; + match master.read(&mut bytes) { + Ok(count) => output.push_str(&String::from_utf8_lossy(&bytes[..count])), + Err(error) if error.raw_os_error() == Some(libc::EIO) => {} + Err(error) => panic!("actual detect: {error}"), + } + } + if !answered && output.contains("\x1b[5n") { + assert!(output.contains("\x1b[16t")); + assert!(output.contains("\x1b_G")); + master + .write_all(b"\x1b_Gi=31;OK\x1b\\\x1b[6;16;8t\x1b[0n") + .unwrap(); + answered = true; + } + if let Some(status) = child.0.try_wait().unwrap() { + assert!( + status.success(), + "actual detect failed ({status}): {output:?}" + ); + assert!( + answered, + "actual detect never queried its controlling tty: {output:?}" + ); + break; + } + } +} diff --git a/src/tui/input.rs b/src/tui/input.rs new file mode 100644 index 0000000..5e01daa --- /dev/null +++ b/src/tui/input.rs @@ -0,0 +1,95 @@ +//! Single-owner terminal input. Create only after capability queries finish; +//! drop (and join) before restoring the terminal or handing stdin to a child. +//! Image/keyboard capability queries may read synchronously during setup, before +//! this owner exists. During its lifetime the UI only consumes the bounded queue: +//! it never polls the OS input reader or acquires crossterm's input lock. +use std::{ + io, + pin::Pin, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, + task::{Context, Poll}, + thread::{self, JoinHandle}, + time::Duration, +}; + +use crossterm::event::{self, Event}; +use futures_util::Stream; +use tokio::sync::mpsc; + +pub(super) struct Events { + receiver: mpsc::Receiver>, + stopped: Arc, + reader: Option>, +} + +impl Events { + pub(super) fn new() -> io::Result { + let (sender, receiver) = mpsc::channel(256); + let stopped = Arc::new(AtomicBool::new(false)); + let reader_stopped = Arc::clone(&stopped); + let reader = thread::Builder::new() + .name("kit-terminal-input".into()) + .spawn(move || { + // No other EventStream or synchronous reader may coexist with this owner. + // poll and read always execute on this same OS thread. + while !reader_stopped.load(Ordering::Acquire) { + let event = match event::poll(Duration::from_millis(25)) { + Ok(false) => continue, + Ok(true) => { + if reader_stopped.load(Ordering::Acquire) { + break; + } + event::read() + } + Err(error) => Err(error), + }; + let failed = event.is_err(); + if reader_stopped.load(Ordering::Acquire) + || sender.blocking_send(event).is_err() + || failed + { + break; + } + } + })?; + Ok(Self { + receiver, + stopped, + reader: Some(reader), + }) + } + + pub(super) fn try_next(&mut self) -> Option> { + self.receiver.try_recv().ok() + } +} + +impl Stream for Events { + type Item = io::Result; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.get_mut().receiver.poll_recv(cx) + } +} + +impl Drop for Events { + fn drop(&mut self) { + // Drop is the sole stop writer. Closing first wakes a sender blocked on + // bounded-channel capacity; the flag bounds an idle reader's next poll. + // Joining hands exclusive terminal ownership back to auth/teardown. + self.receiver.close(); + self.stopped.store(true, Ordering::Release); + if let Some(reader) = self.reader.take() { + // A worker panic is already reported by the panic hook and closes + // the stream. Never turn cleanup (possibly unwinding) into a panic. + let _ = reader.join(); + } + } +} + +#[cfg(all(test, unix))] +#[path = "input_tests.rs"] +mod tests; diff --git a/src/tui/input_tests.rs b/src/tui/input_tests.rs new file mode 100644 index 0000000..ea98b17 --- /dev/null +++ b/src/tui/input_tests.rs @@ -0,0 +1,331 @@ +//! Real PTYs isolate crossterm's process-global decoder. No production hooks. +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::disallowed_methods, + clippy::disallowed_macros +)] + +use super::Events; +use crate::tui::{enter, leave, resume_terminal}; +use futures_util::FutureExt; +use std::{ + fs::File, + io::{Read, Write}, + os::{ + fd::{AsRawFd, FromRawFd}, + unix::process::CommandExt, + }, + process::{Child, Command, Stdio}, + time::{Duration, Instant}, +}; + +const CHILD_ENV: &str = "KIT_INPUT_LIFECYCLE_CHILD"; +const TEST_NAME: &str = "tui::input::tests::terminal_input_lifecycle"; + +struct TestChild(Child); +impl Drop for TestChild { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn marker(value: &str) { + println!("INPUT_TEST:{value}"); + std::io::stdout().flush().unwrap(); +} + +fn wait_until(mut ready: impl FnMut() -> bool) { + let deadline = Instant::now() + Duration::from_secs(5); + while !ready() { + assert!(Instant::now() < deadline, "input condition timed out"); + std::thread::sleep(Duration::from_millis(5)); + } +} + +fn key(events: &mut Events, expected: char) { + wait_until(|| match events.try_next() { + Some(Ok(crossterm::event::Event::Key(key))) => { + assert_eq!(key.code, crossterm::event::KeyCode::Char(expected)); + true + } + Some(Err(error)) => panic!("input failed: {error}"), + _ => false, + }); +} + +fn auth_child(cancel: bool) { + assert!(!crossterm::terminal::is_raw_mode_enabled().unwrap()); + // Exercise the real authentication wait boundary with inherited PTY stdin. + // The child signals Stop only after it has received its complete line. + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let mut stop = crate::tui::Stop::new().unwrap(); + let mut command = tokio::process::Command::new("/bin/sh"); + command.args(["-c", if cancel { + "printf 'INPUT_TEST:AUTH\\n'; IFS= read -r line; test \"$line\" = auth || exit 42; kill -TERM \"$PPID\"; exec sleep 30" + } else { + "printf 'INPUT_TEST:AUTH\\n'; IFS= read -r line; test \"$line\" = auth || exit 42; exit 17" + }]); + let result = crate::tui::wait_for_terminal_auth(command, &mut stop).await; + if cancel { + assert!(result.is_none()); + } else { + assert_eq!(result.unwrap().unwrap().code(), Some(17)); + } + }); +} + +fn child_case(case: &str) { + match case { + "idle" | "saturated" | "handoff" | "handoff_cancel" | "handoff_unanswered" + | "handoff_partial" => { + let (mut terminal, _images) = enter().unwrap(); + let mut events = Events::new().unwrap(); + if case == "saturated" { + marker("PRESSURE"); + // Observe the real bounded queue reaching capacity, rather than + // assuming that sleeping or writing N bytes implies saturation. + wait_until(|| events.receiver.len() == events.receiver.max_capacity()); + } else if case.starts_with("handoff") { + marker("FIRST"); + key(&mut events, 'x'); + } + drop(events); + leave(&mut terminal); + if case.starts_with("handoff") { + auth_child(case == "handoff_cancel"); + let _images = resume_terminal(&mut terminal).unwrap(); + let mut events = Events::new().unwrap(); + marker("RESTART"); + key(&mut events, 'z'); + drop(events); + } + drop(terminal); + } + "cancel" => { + // Poll through actual terminal setup and input acquisition, suspend + // at an await, then cancel by dropping the owning future. + let future = async { + let (_terminal, _images) = enter().unwrap(); + let _events = Events::new().unwrap(); + std::future::pending::<()>().await; + }; + assert!(future.now_or_never().is_none()); + auth_child(false); + } + "unwind" => { + let result = std::panic::catch_unwind(|| { + let (_terminal, _images) = enter().unwrap(); + let _events = Events::new().unwrap(); + panic!("unwind while owning input"); + }); + assert!(result.is_err()); + auth_child(false); + } + "startup_no_cursor" => { + let (mut terminal, _images) = enter().unwrap(); + let mut events = Events::new().unwrap(); + marker("FIRST"); + key(&mut events, 'x'); + drop(events); + leave(&mut terminal); + auth_child(false); + } + "worker_unwind" => { + let (mut terminal, _images) = enter().unwrap(); + let mut events = Events::new().unwrap(); + // A different owner's panic must not restore our live terminal. + assert!( + std::thread::spawn(|| panic!("unrelated worker panic")) + .join() + .is_err() + ); + assert!(crossterm::terminal::is_raw_mode_enabled().unwrap()); + marker("FIRST"); + key(&mut events, 'x'); + drop(events); + leave(&mut terminal); + auth_child(false); + } + "setup_failure" => { + let (mut terminal, _images) = enter().unwrap(); + leave(&mut terminal); + let saved = std::io::stdout().as_raw_fd(); + // A genuine output failure after raw mode is enabled exercises + // rollback at the terminal boundary, on every Unix platform. + let copy = unsafe { libc::dup(saved) }; + assert!(copy >= 0); + let readonly = File::open("/dev/null").unwrap(); + assert_eq!(unsafe { libc::dup2(readonly.as_raw_fd(), saved) }, saved); + let result = resume_terminal(&mut terminal); + assert_eq!(unsafe { libc::dup2(copy, saved) }, saved); + unsafe { + libc::close(copy); + } + assert!(result.is_err()); + drop(terminal); + // Check kernel state, not just crossterm's cached raw-mode state. + let mut modes = std::mem::MaybeUninit::::uninit(); + // SAFETY: stdin is this child's PTY and modes is writable. + assert_eq!(unsafe { libc::tcgetattr(0, modes.as_mut_ptr()) }, 0); + let modes = unsafe { modes.assume_init() }; + assert_ne!(modes.c_lflag & libc::ICANON, 0); + assert_ne!(modes.c_lflag & libc::ECHO, 0); + auth_child(false); + // A failed setup must not leave a reader that can steal the next + // interval's input either. Re-enter from restored kernel modes. + let (terminal, _images) = enter().unwrap(); + let mut events = Events::new().unwrap(); + marker("RESTART"); + key(&mut events, 'z'); + drop(events); + drop(terminal); + } + _ => panic!("unknown case"), + } + assert!(!crossterm::terminal::is_raw_mode_enabled().unwrap()); + marker("DONE"); +} + +#[test] +fn terminal_input_lifecycle() { + if let Ok(case) = std::env::var(CHILD_ENV) { + child_case(&case); + return; + } + for case in [ + "idle", + "startup_no_cursor", + "saturated", + "handoff", + "handoff_cancel", + "handoff_unanswered", + "handoff_partial", + "cancel", + "unwind", + "worker_unwind", + "setup_failure", + ] { + run_pty(case); + } +} + +fn run_pty(case: &str) { + let (mut master_fd, mut slave_fd) = (-1, -1); + // SAFETY: openpty initializes two distinct owned descriptors. + assert_eq!( + unsafe { + libc::openpty( + &mut master_fd, + &mut slave_fd, + std::ptr::null_mut(), + std::ptr::null_mut(), + std::ptr::null_mut(), + ) + }, + 0 + ); + let (mut master, slave) = + unsafe { (File::from_raw_fd(master_fd), File::from_raw_fd(slave_fd)) }; + let mut command = Command::new(std::env::current_exe().unwrap()); + command + .args(["--exact", TEST_NAME, "--nocapture"]) + .env(CHILD_ENV, case) + .stdin(Stdio::from(slave.try_clone().unwrap())) + .stdout(Stdio::from(slave.try_clone().unwrap())) + .stderr(Stdio::from(slave)); + // SAFETY: only async-signal-safe calls between fork and exec. Queries and + // reads use the child's controlling PTY, never the developer's terminal. + unsafe { + command.pre_exec(|| { + if libc::setsid() == -1 || libc::ioctl(0, libc::TIOCSCTTY as _, 0) == -1 { + return Err(std::io::Error::last_os_error()); + } + Ok(()) + }); + } + let mut child = TestChild(command.spawn().unwrap()); + drop(command); + let deadline = Instant::now() + Duration::from_secs(20); + let mut output = String::new(); + let mut queries = 0; + let mut cursor_queries = 0; + let mut status_queries = 0; + let mut sent = Vec::new(); + loop { + assert!(Instant::now() < deadline, "{case} timed out: {output:?}"); + let mut fd = libc::pollfd { + fd: master.as_raw_fd(), + events: libc::POLLIN, + revents: 0, + }; + // SAFETY: fd points to one initialized pollfd. + if unsafe { libc::poll(&mut fd, 1, 20) } > 0 { + let mut bytes = [0; 4096]; + match master.read(&mut bytes) { + Ok(count) => output.push_str(&String::from_utf8_lossy(&bytes[..count])), + Err(error) if error.raw_os_error() == Some(libc::EIO) => {} + Err(error) => panic!("{case}: {error}"), + } + } + // Timeout cases deliberately never complete the image query. Both + // initial entry and resume must finish it before accepting keyboard + // input or handing stdin to an authentication child. + let status_count = output.matches("\x1b[5n").count(); + while status_queries < status_count { + match case { + "handoff_unanswered" => {} + "handoff_partial" => master.write_all(b"\x1b[4;").unwrap(), + _ => master.write_all(b"\x1b[0n").unwrap(), + } + status_queries += 1; + } + let cursor_count = output.matches("\x1b[6n").count(); + if case == "startup_no_cursor" { + assert_eq!( + cursor_count, 0, + "startup introduced a cursor-report requirement" + ); + } + while cursor_queries < cursor_count { + master.write_all(b"\x1b[1;1R").unwrap(); + cursor_queries += 1; + } + let query_count = output.matches("\x1b[?u").count(); + while queries < query_count { + master.write_all(b"\x1b[?1u\x1b[?1;2c").unwrap(); + queries += 1; + } + for (name, bytes) in [ + ("PRESSURE", vec![b'a'; 1024]), + ("FIRST", b"x".to_vec()), + ("AUTH", b"auth\n".to_vec()), + ("RESTART", b"z".to_vec()), + ] { + if !sent.contains(&name) && output.contains(&format!("INPUT_TEST:{name}")) { + master.write_all(&bytes).unwrap(); + sent.push(name); + } + } + if let Some(status) = child.0.try_wait().unwrap() { + assert!(status.success(), "{case} failed ({status}): {output:?}"); + assert!(output.contains("INPUT_TEST:DONE"), "{case}: {output:?}"); + let expected_queries = if case.starts_with("handoff") || case == "setup_failure" { + 2 + } else { + 1 + }; + assert!( + status_queries >= expected_queries, + "{case}: actual image detection skipped its queries: {output:?}" + ); + break; + } + } +} diff --git a/src/tui/mod.rs b/src/tui/mod.rs index 5fe3162..2a0c4a4 100644 --- a/src/tui/mod.rs +++ b/src/tui/mod.rs @@ -12,6 +12,7 @@ mod command; mod editor; mod hyperlinks; mod image; +mod input; #[cfg(all(test, unix))] mod keyboard_tests; mod markdown; @@ -45,8 +46,8 @@ use base64::{Engine as _, engine::general_purpose::STANDARD}; use crossterm::{ event::{ DisableBracketedPaste, DisableMouseCapture, EnableBracketedPaste, EnableMouseCapture, - Event, EventStream, KeyCode, KeyEvent, KeyEventKind, KeyModifiers, - KeyboardEnhancementFlags, PopKeyboardEnhancementFlags, PushKeyboardEnhancementFlags, + Event, KeyCode, KeyEvent, KeyEventKind, KeyModifiers, KeyboardEnhancementFlags, + PopKeyboardEnhancementFlags, PushKeyboardEnhancementFlags, }, execute, style::Print, @@ -1637,7 +1638,8 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( initialized.capabilities.session.as_ref().and_then(|session| session.inject.as_ref()), ); app.auth_methods = auth_methods; - let mut events = EventStream::new(); + // TerminalSession restores modes if spawning the reader fails. + let mut events = input::Events::new().map_err(agent_client_protocol::Error::into_internal_error)?; let mut ticker = tokio::time::interval(TICK); ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); @@ -1656,6 +1658,7 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( let mut next_priority = 0; loop { if let Err(error) = draw_frame(&mut terminal, &mut app, &mut images) { + drop(events); leave(&mut terminal); return Err(agent_client_protocol::Error::into_internal_error(error)); } @@ -1688,13 +1691,20 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( Err(LoginEvent::Terminal(event)) => { let action = match event { Some(Ok(event)) => handle(&mut app, event), - Some(Err(_)) | None => { + Some(Err(error)) => { + drop(events); + leave(&mut terminal); + return Err(agent_client_protocol::Error::into_internal_error(error)); + } + None => { + drop(events); leave(&mut terminal); return Ok(()); } }; match action { Action::Quit => { + drop(events); leave(&mut terminal); return Ok(()); } @@ -1769,7 +1779,7 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( images = resume_terminal(&mut terminal).map_err( agent_client_protocol::Error::into_internal_error, )?; - events = EventStream::new(); + events = input::Events::new().map_err(agent_client_protocol::Error::into_internal_error)?; break session; } Err(error) if authentication_required( @@ -1794,7 +1804,7 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( images = resume_terminal(&mut terminal).map_err( agent_client_protocol::Error::into_internal_error, )?; - events = EventStream::new(); + events = input::Events::new().map_err(agent_client_protocol::Error::into_internal_error)?; } Action::None | Action::Redraw => {} Action::ReadClipboard(_, _) => { @@ -1806,12 +1816,14 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( Ok(update) => match update { Some(update) => app.apply(update.update), None => { + drop(events); leave(&mut terminal); return Ok(()); } }, Err(LoginEvent::Tick) => app.tick(), Err(LoginEvent::Stop) => { + drop(events); leave(&mut terminal); return Ok(()); } @@ -1822,6 +1834,7 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( let active_session_id = match durable_session_id(&session_id) { Ok(session_id) => session_id, Err(error) => { + drop(events); leave(&mut terminal); return Err(agent_client_protocol::Error::into_internal_error( std::io::Error::other(error), @@ -1841,6 +1854,9 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( app.start_session(active_session_id.clone()); let storage_shutdown = crate::resilient_fs::shutdown_token(); let mut voice = NativeVoice::default(); + // This scope owns events (authentication moves/drops it) but only + // borrows terminal. Cancelling it joins input before the outer + // TerminalSession guard restores modes, just like ordinary return. let result: Result<(), agent_client_protocol::Error> = async { enum SessionEvent { @@ -2002,18 +2018,14 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( for index in 0..MAX_BURST { match next { Some(Ok(event)) => action = handle_with_clipboard(&mut app, &mut clipboard_pastes, event), - Some(Err(_)) | None => return Ok(()), + Some(Err(error)) => return Err(agent_client_protocol::Error::into_internal_error(error)), + None => return Ok(()), } if !matches!(action, Action::None) || index + 1 == MAX_BURST { break; } - // `EventStream::next().now_or_never()` polls with a noop - // waker. If no event is ready, crossterm's background reader - // retains that waker and cannot wake this select loop when the - // next key arrives. Check synchronously before polling the - // stream so an empty burst cannot make the TUI unresponsive. - if crossterm::event::poll(Duration::ZERO).unwrap_or(false) { - next = events.next().await; + if let Some(event) = events.try_next() { + next = Some(event); } else { break; } @@ -2457,7 +2469,7 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( images = resume_terminal(&mut terminal).map_err( agent_client_protocol::Error::into_internal_error, )?; - events = EventStream::new(); + events = input::Events::new().map_err(agent_client_protocol::Error::into_internal_error)?; } Action::OpenUserImage(image) => { // Snapshot only; release the guard before queueing work. @@ -2630,6 +2642,7 @@ pub async fn run_with_reasoning_effort_and_openrouter_key( } } .await; + // The completed async block owned and dropped the input reader. // Best effort, nonblocking even on quit or an early error. Audio // stops first; connection teardown is the final cleanup boundary. voice.stop(); @@ -3263,39 +3276,85 @@ fn draw_frame( }) } -fn enter() -> std::io::Result<(DefaultTerminal, image::ImageRuntime)> { - TERMINAL_ACTIVE.store(true, Ordering::Relaxed); - let previous = std::panic::take_hook(); - std::panic::set_hook(Box::new(move |info| { - restore_modes(); - ratatui::restore(); - previous(info); - })); - crossterm::terminal::enable_raw_mode()?; - execute!(std::io::stdout(), EnterAlternateScreen)?; - let terminal = ratatui::Terminal::new(hyperlinks::HyperlinkBackend::new(std::io::stdout()))?; - // Query after entering the alternate screen but before the event stream owns - // terminal input, as required by ratatui-image. The query has a short bound. - let images = image::ImageRuntime::detect(); - // Bracketed paste keeps pasted newlines out of the key stream. Keyboard - // enhancement distinguishes command keys, shifted returns, and releases. - enable_tui_modes(); +/// Owns terminal modes across fallible setup, dropped futures, and unwind. +/// Declare input after this guard so its reader joins before mode restoration. +struct TerminalSession { + terminal: DefaultTerminal, + active: bool, +} + +impl std::ops::Deref for TerminalSession { + type Target = DefaultTerminal; + fn deref(&self) -> &Self::Target { + &self.terminal + } +} + +impl std::ops::DerefMut for TerminalSession { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.terminal + } +} + +impl Drop for TerminalSession { + fn drop(&mut self) { + if self.active { + leave(self); + } + } +} + +fn enter() -> std::io::Result<(TerminalSession, image::ImageRuntime)> { + // Unwinding restores through TerminalSession, after Events has joined. + // A process-global unwind hook would restore too early, including for an + // unrelated worker panic whose terminal owner remains alive. + #[cfg(panic = "abort")] + { + // Abort cannot run either destructor. Preserve best-effort emergency + // restoration and diagnostic visibility, without claiming joined input. + let previous = std::panic::take_hook(); + std::panic::set_hook(Box::new(move |info| { + restore_modes(); + ratatui::restore(); + previous(info); + })); + } + // Construct before changing modes so a construction failure needs no cleanup. + let mut terminal = TerminalSession { + terminal: ratatui::Terminal::new(hyperlinks::HyperlinkBackend::new(std::io::stdout()))?, + active: false, + }; + let images = prepare_terminal(&mut terminal)?; Ok((terminal, images)) } -fn resume_terminal(terminal: &mut DefaultTerminal) -> std::io::Result { +fn resume_terminal(terminal: &mut TerminalSession) -> std::io::Result { + let images = prepare_terminal(terminal)?; + if let Err(error) = terminal.clear() { + leave(terminal); + return Err(error); + } + Ok(images) +} + +fn prepare_terminal(terminal: &mut TerminalSession) -> std::io::Result { + terminal.active = true; TERMINAL_ACTIVE.store(true, Ordering::Relaxed); let resumed = (|| { crossterm::terminal::enable_raw_mode()?; execute!(std::io::stdout(), EnterAlternateScreen)?; + // Image and synchronous keyboard-enhancement capability queries are + // setup-only readers. Both must finish before Events acquires input; + // steady-state UI code consumes only its queue, never crossterm locks. let images = image::ImageRuntime::detect(); enable_tui_modes(); - terminal.clear()?; + // Initial entry has fresh Ratatui buffers and a fresh alternate screen. + // Only resume needs clear(): it also queries the cursor position, which + // ordinary startup must not require the terminal to report. Ok(images) })(); if resumed.is_err() { - restore_modes(); - ratatui::restore(); + leave(terminal); } resumed } @@ -3437,7 +3496,8 @@ fn print_exit_message(app: &App) { let _ = stdout.flush(); } -fn leave(terminal: &mut DefaultTerminal) { +fn leave(terminal: &mut TerminalSession) { + terminal.active = false; restore_modes(); let _ = terminal.show_cursor(); ratatui::restore(); diff --git a/src/tui/startup.rs b/src/tui/startup.rs index ee49a96..4b87e70 100644 --- a/src/tui/startup.rs +++ b/src/tui/startup.rs @@ -2,7 +2,7 @@ use std::path::Path; -use crossterm::event::{Event, EventStream, KeyCode, KeyEventKind, KeyModifiers}; +use crossterm::event::{Event, KeyCode, KeyEventKind, KeyModifiers}; use futures_util::{ StreamExt, future::{Either, select}, @@ -47,8 +47,10 @@ pub async fn pick_session( // terminal on both successful cancellation and fallible reads/draws. let (mut terminal, mut images) = enter()?; let mut renames = RenameCommits::default(); + // The awaited scope owns input but borrows the terminal guard: on normal + // return, error, or cancellation it joins input before terminal restoration. let result = async { - let mut events = EventStream::new(); + let mut events = super::input::Events::new()?; let mut clipboard_pending = false; let mut ticker = tokio::time::interval(super::TICK); ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); diff --git a/tests/support/terminal_latency/README.md b/tests/support/terminal_latency/README.md new file mode 100644 index 0000000..93746ed --- /dev/null +++ b/tests/support/terminal_latency/README.md @@ -0,0 +1,72 @@ +# Real TUI terminal-output latency probe + +This opt-in probe uses only existing Rust dependencies and Python's standard +library. It calls the actual public TUI entry point, including crossterm input, +the real event loop, HyperlinkBackend, ANSI output, and a controlling PTY. The +TUI's normal `current_exe() serve ...` child launch is dispatched by a test-only +executable to a local ACP v2 fixture. There is no provider, credential, or network +service. No production source, manifest, or stored session needs modification. + +This is **key write → arrival of rendered glyph bytes and their synchronized-frame end at the PTY master**, +including scheduler and reader overhead. It is not terminal-emulator paint or +screen-photon latency. It bypasses CLI configuration parsing and the real ACP +server/provider, not the production TUI loop. It is not a TestBackend benchmark. + +## Run + +Use detached temporary worktrees for each revision. Copy `driver.rs` into each +worktree as `examples/terminal_latency_probe.rs` (Cargo auto-discovers examples). +In that worktree, use the pinned toolchain: + +```sh +mise run build:release -- --example terminal_latency_probe +python3 /absolute/path/to/tests/support/terminal_latency/probe.py \ + /absolute/path/to/target/release/examples/terminal_latency_probe \ + --output /tmp/latency-current --samples 100 --history 1000 +``` + +Trust the temporary worktree's mise configuration if required. To reuse build +artifacts, set `CARGO_TARGET_DIR` to an existing target directory, build revisions +sequentially, and copy each finished example executable to a revision-specific +path before building the next. **The build task also rebuilds `target/release/kit`; +a shared target can therefore leave that binary at a baseline or experimental +revision. After all timed comparisons, run `mise run build:release` from the +original checkout to restore its binary. Do not replace an installed binary or +restart live sessions for this probe.** The executable uses the probe script supplied by +the runner, so both revisions execute the same fixture. Never run comparative +measurements concurrently with compilation or another measurement. + +The runner isolates HOME and cwd in a temporary directory and constructs an +environment allowlist. It asserts raw/no-echo terminal mode to reject accidental +kernel-echo measurements. The ACP fixture contains only generated text. Idle +samples alternate Q/Z insertion and backspace. The fixture never emits these +uppercase glyphs. Hot samples begin 100 ms after submit, during a deterministic burst of `--history` distinct markdown messages, then +one markdown chunk every 5 ms. Both paths use a 120×40 PTY. Each insertion waits +for its corresponding printable output bytes and the following synchronized-output +end marker (`CSI ? 2026 l`); terminal escape sequences are +removed before matching. A 30 ms drain follows each deletion. This is deliberately +paced typing, not a saturated keyboard-throughput test. + +`summary.json` records the binary SHA-256, nearest-rank p50/p95/p99/max, and +per-mode elapsed time, output bytes, and completed synchronized frames. +`events.jsonl` retains monotonic +key/read timestamps and individual latency samples; `terminal.bin` retains raw +ANSI output. `agent-requests.jsonl` contains only requests to the synthetic fixture. +The runner also requires that streaming text was actually rendered during the hot run. A timeout is a failure, not a dropped sample. Keep the logs local. +The percentile includes transport and Python scheduling delay and is an upper +bound on the backend's write completion; there is no physical renderer in this +probe. Repeat runs and vary history depth before drawing causal conclusions. + +## Event-reader contention + +Each binary executes its revision's production event-reader architecture. +Comparisons with older revisions can expose the effect of replacing EventStream +and synchronous UI-thread polling with a dedicated reader/channel. A revision +comparison alone does not isolate reader effects from other changes: use a +separately labeled, single-change variant when investigating causality. + +The support checks can run without launching a TUI: + +```sh +python3 -m unittest discover -s tests/support/terminal_latency -p 'test_*.py' +``` diff --git a/tests/support/terminal_latency/driver.rs b/tests/support/terminal_latency/driver.rs new file mode 100644 index 0000000..95db979 --- /dev/null +++ b/tests/support/terminal_latency/driver.rs @@ -0,0 +1,26 @@ +//! Test-only entry point: real TUI, fake ACP peer at the process boundary. +use std::{os::unix::process::CommandExt, path::Path}; + +#[tokio::main] +async fn main() -> Result<(), Box> { + if std::env::args().nth(1).as_deref() == Some("serve") { + return Err(std::process::Command::new(std::env::var("PROBE_PYTHON")?) + .arg(std::env::var("PROBE_SCRIPT")?) + .arg("--agent") + .args(std::env::args().skip(2)) + .exec() + .into()); + } + kit::tui::run( + Path::new(&std::env::var("PROBE_ROOT")?), + "latency-fixture", + kit::ProviderKind::OpenRouter, + None, + None, + &Default::default(), + &Default::default(), + None, + false, + ) + .await +} diff --git a/tests/support/terminal_latency/probe.py b/tests/support/terminal_latency/probe.py new file mode 100644 index 0000000..12ce37e --- /dev/null +++ b/tests/support/terminal_latency/probe.py @@ -0,0 +1,195 @@ +#!/usr/bin/env python3 +"""Real TUI PTY probe; stdlib only. See README.md for scope and reproduction.""" +import argparse +import errno +import fcntl +import hashlib +import json +import math +import os +from pathlib import Path +import pty +import re +import select +import signal +import struct +import sys +import tempfile +import termios +import time + + +def agent(): + def send(message): + print(json.dumps(message), flush=True) + + def update(session, index, text): + send({"jsonrpc": "2.0", "method": "session/update", "params": { + "sessionId": session, "update": {"sessionUpdate": "agent_message_chunk", + "messageId": f"message-{index}", "content": {"type": "text", "text": text}}}}) + + for line in sys.stdin: + request = json.loads(line) + method = request.get("method") + with open(os.environ["PROBE_REQUEST_LOG"], "a") as log: + log.write(json.dumps(request) + "\n") + result = {} + if method == "initialize": + result = {"protocolVersion": 2, "info": {"name": "latency-fixture", "version": "1"}, + "capabilities": {"session": {}}, "authMethods": []} + elif method == "session/new": + session = sys.argv[sys.argv.index("--session-id") + 1] + result = {"sessionId": session} + elif method == "session/prompt": + session = request["params"]["sessionId"] + send({"jsonrpc": "2.0", "id": request["id"], "result": {}}) + send({"jsonrpc": "2.0", "method": "session/update", "params": { + "sessionId": session, "update": {"sessionUpdate": "state_update", "state": "running"}}}) + # Finite deterministic transcript prefix, then paced streaming for 30 seconds. + for index in range(int(os.environ.get("PROBE_HISTORY", "1000"))): + update(session, index, f"replay {index}: **bold** `code` and a small paragraph.\n\n") + for index in range(6000): + update(session, 100000, f"stream {index}: some text with **markdown**.\n") + time.sleep(0.005) + send({"jsonrpc": "2.0", "method": "session/update", "params": { + "sessionId": session, "update": {"sessionUpdate": "state_update", "state": "idle", + "stopReason": "end_turn"}}}) + continue + elif method is None: + continue + if "id" in request: + send({"jsonrpc": "2.0", "id": request["id"], "result": result}) + + +QUERIES = [ + (b"\x1b_Gi=31,s=1,v=1,a=q,t=d,f=24;AAAA\x1b\\", b"\x1b_Gi=31;ENOTSUP\x1b\\"), + (b"\x1b[16t", b"\x1b[6;16;8t"), (b"\x1b[5n", b"\x1b[0n"), + (b"\x1b[?u", b"\x1b[?0u"), (b"\x1b[c", b"\x1b[?1;2c"), + (b"\x1b[6n", b"\x1b[1;1R"), +] +# Strip terminal commands before looking for the deliberately unique uppercase glyph. +ESCAPES = re.compile(rb"\x1b\].*?(?:\x07|\x1b\\)|\x1b\[[0-?]*[ -/]*[@-~]|\x1b[()][A-Z0-9]|\x1b.") + + +def run(args): + args.binary = str(Path(args.binary).resolve()) + if args.samples < 1 or args.history < 0: + raise ValueError("samples must be positive; history must be nonnegative") + output = Path(args.output).resolve() + output.mkdir(parents=True, exist_ok=True) + with tempfile.TemporaryDirectory(prefix="kit-latency-") as directory: + home = Path(directory) + root = home / "workspace" + root.mkdir() + pid, master = pty.fork() + if pid == 0: + env = {"PATH": os.environ["PATH"], "HOME": str(home), "TERM": "xterm-256color", + "LANG": "en_US.UTF-8", "PROBE_ROOT": str(root), "PROBE_SCRIPT": str(Path(__file__).resolve()), + "PROBE_REQUEST_LOG": str(output / "agent-requests.jsonl"), "PROBE_PYTHON": sys.executable, "PROBE_HISTORY": str(args.history)} + os.chdir(root) + os.execve(str(Path(args.binary).resolve()), [args.binary], env) + fcntl.ioctl(master, termios.TIOCSWINSZ, struct.pack("HHHH", 40, 120, 0, 0)) + pending = b"" + frame_end = b"\x1b[?2026l" + frame_tail = b"" + output_bytes = 0 + completed_frames = 0 + raw = open(output / "terminal.bin", "wb") + events = open(output / "events.jsonl", "w") + def pump(timeout): + nonlocal pending, frame_tail, output_bytes, completed_frames + if not select.select([master], [], [], timeout)[0]: + return b"" + try: + data = os.read(master, 65536) + except OSError as error: + if error.errno == errno.EIO: + raise RuntimeError("TUI exited; inspect terminal.bin") from error + raise + if not data: + raise RuntimeError("TUI exited") + raw.write(data) + output_bytes += len(data) + framed = frame_tail + data + completed_frames += framed.count(frame_end) + frame_tail = framed[-(len(frame_end) - 1):] + events.write(json.dumps({"read_ns": time.perf_counter_ns(), "bytes": len(data)}) + "\n") + pending += data + for query, reply in QUERIES: + while query in pending: + os.write(master, reply) + pending = pending.replace(query, b"", 1) + pending = pending[-256:] + return data + def drain(seconds): + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + pump(min(0.01, max(0, deadline-time.monotonic()))) + def samples(mode): + mode_started = time.monotonic() + initial_bytes, initial_frames = output_bytes, completed_frames + values = [] + for index in range(args.samples): + drain(0.03) + glyph = b"Q" if index % 2 == 0 else b"Z" + started = time.perf_counter_ns() + os.write(master, glyph) + received = b"" + deadline = time.monotonic() + 10 + while (glyph not in ESCAPES.sub(b"", received) or + b"\x1b[?2026l" not in received[received.rfind(glyph) + 1:]): + if time.monotonic() > deadline: + raise TimeoutError(f"no rendered glyph in {mode} sample {index}") + received += pump(0.1) + ended = time.perf_counter_ns() + latency = (ended-started)/1e6 + values.append(latency) + events.write(json.dumps({"mode": mode, "sample": index, "key_ns": started, + "frame_end_ns": ended, "latency_ms": latency}) + "\n") + os.write(master, b"\x7f") + ordered = sorted(values) + return {"n": len(values), **{f"p{p}_ms": ordered[math.ceil(p/100*len(values))-1] + for p in (50, 95, 99)}, "max_ms": max(values), + "elapsed_seconds": time.monotonic() - mode_started, + "output_bytes": output_bytes - initial_bytes, + "completed_frames": completed_frames - initial_frames} + try: + drain(3) + if termios.tcgetattr(master)[3] & (termios.ECHO | termios.ICANON): + raise RuntimeError("TUI did not enter raw mode; refusing to measure PTY echo") + idle = samples("idle") + drain(0.2) + os.write(master, b"replay") + drain(0.2) + os.write(master, b"\r") + drain(0.1) + hot = samples("hot") + raw.flush() + if b"stream" not in (output / "terminal.bin").read_bytes(): + raise RuntimeError("no replay stream was rendered during hot measurements") + digest = hashlib.sha256(Path(args.binary).read_bytes()).hexdigest() + result = {"binary": args.binary, "binary_sha256": digest, "history_messages": args.history, + "terminal": "120x40", "idle": idle, "hot": hot} + (output / "summary.json").write_text(json.dumps(result, indent=2) + "\n") + print(json.dumps(result)) + finally: + try: + os.kill(pid, signal.SIGKILL) + except ProcessLookupError: + pass + os.close(master) + os.waitpid(pid, 0) + raw.close() + events.close() + + +if __name__ == "__main__": + if "--agent" in sys.argv: + agent() + else: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("binary") + parser.add_argument("--output", required=True) + parser.add_argument("--samples", type=int, default=100) + parser.add_argument("--history", type=int, default=1000) + run(parser.parse_args()) diff --git a/tests/support/terminal_latency/test_probe.py b/tests/support/terminal_latency/test_probe.py new file mode 100644 index 0000000..15fb2ff --- /dev/null +++ b/tests/support/terminal_latency/test_probe.py @@ -0,0 +1,45 @@ +"""Small protocol/measurement checks; the actual latency probe still requires a PTY.""" +import importlib.util +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest + +SCRIPT = Path(__file__).with_name("probe.py") +spec = importlib.util.spec_from_file_location("probe", SCRIPT) +probe = importlib.util.module_from_spec(spec) +spec.loader.exec_module(probe) + + +class ProbeTests(unittest.TestCase): + def test_escape_stripping_preserves_glyph_after_closed_hyperlink(self): + self.assertEqual(probe.ESCAPES.sub(b"", b"\x1b]8;;\x1b\\Q\x1b[?2026l"), b"Q") + + def test_v2_prompt_is_acknowledged_before_streaming(self): + with tempfile.TemporaryDirectory() as directory: + env = dict(os.environ, PROBE_REQUEST_LOG=str(Path(directory) / "requests.jsonl"), + PROBE_HISTORY="1") + child = subprocess.Popen([sys.executable, str(SCRIPT), "--agent", "--session-id", "test"], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, + text=True, env=env) + try: + child.stdin.write(json.dumps({"jsonrpc": "2.0", "id": "p", "method": "session/prompt", + "params": {"sessionId": "test", "prompt": []}}) + "\n") + child.stdin.flush() + self.assertEqual(json.loads(child.stdout.readline()), + {"jsonrpc": "2.0", "id": "p", "result": {}}) + state = json.loads(child.stdout.readline()) + self.assertEqual(state["params"]["update"], + {"sessionUpdate": "state_update", "state": "running"}) + chunk = json.loads(child.stdout.readline()) + self.assertEqual(chunk["params"]["update"]["sessionUpdate"], "agent_message_chunk") + finally: + child.kill() + child.communicate(timeout=5) + + +if __name__ == "__main__": + unittest.main()