jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-04 16:31:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

merge: pin mounted terminal to blessed runtime floor

Browse Source 
Co-authored-by: npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta <d8473ee32b973aa31a21a65adddcc4b69cc2a8a4dee8121ecd51926e0cddbc02@buzz.block.builderlab.xyz>
Signed-off-by: npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta <d8473ee32b973aa31a21a65adddcc4b69cc2a8a4dee8121ecd51926e0cddbc02@buzz.block.builderlab.xyz>
This commit is contained in:
npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta
2026-08-01 22:22:11 -04:00
parent 407db04792 4bd3b03a3c
commit 8718137ece
15 changed files with 2722 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
desktop/src-tauri/Cargo.lock generated
View File

@@ -1073,6 +1073,7 @@ dependencies = [
"buzz-media",
"buzz-persona",
"buzz-sdk",
"buzz-terminal",
"buzz-voice",
"bytes",
"bzip2 0.6.1",
@@ -1105,6 +1106,7 @@ dependencies = [
"opus",
"plist",
"png 0.18.1",
"portable-pty",
"regex",
"reqwest 0.13.4",
"rodio",

2
desktop/src-tauri/Cargo.toml
View File

@@ -104,6 +104,8 @@ buzz_persona_pkg = { package = "buzz-persona", path = "../../crates/buzz-persona
buzz_sdk_pkg = { package = "buzz-sdk", path = "../../crates/buzz-sdk" }
buzz_agent_pkg = { package = "buzz-agent", path = "../../crates/buzz-agent" }
buzz_voice_pkg = { package = "buzz-voice", path = "../../crates/buzz-voice" }
buzz_terminal = { package = "buzz-terminal", path = "crates/buzz-terminal" }
portable-pty = "0.9"
iroh = { version = "1.0.2", optional = true }
mesh-llm-sdk = { git = "https://github.com/Mesh-LLM/mesh-llm.git", tag = "v0.74.0", package = "mesh-llm-sdk", default-features = false, features = ["client", "serving"], optional = true }
mesh-llm-host-runtime = { git = "https://github.com/Mesh-LLM/mesh-llm.git", tag = "v0.74.0", package = "mesh-llm-host-runtime", default-features = false, features = ["dynamic-native-runtime"], optional = true }

171
desktop/src-tauri/crates/buzz-terminal/src/damage.rs
View File

@@ -45,7 +45,31 @@ use alacritty_terminal::index::{Column, Line};
use alacritty_terminal::term::cell::{Cell, Flags};
use alacritty_terminal::term::TermDamage;
/// A run of cells sharing one visual style.
/// A run of cells sharing one visual style **and one cell width**.
///
/// # Why the consumer can position every cluster without Unicode tables
///
/// The renderer must place each display cluster at its true column, and it
/// cannot derive that from the text: no single split rule over a concatenated
/// string is correct. A regional-indicator flag (`U+1F1FA U+1F1F8`) is two
/// ordinary one-column cells, so it must split *per codepoint*; a keycap
/// (`1 U+FE0F U+20E3`) is one cell holding three codepoints, so it must split
/// *per grapheme*. Those rules disagree, and the distinction lives in the grid,
/// not in the string.
///
/// So the run carries it instead. Within a span every cluster advances the same
/// [`width`](Self::width) columns, and [`cluster_count`](Self::cluster_count)
/// says how many clusters the text holds. The consumer's rule is arithmetic on
/// those two numbers, with no Unicode table anywhere:
///
/// ```text
/// cluster_count == 1 -> the whole text is one cluster, at `column`
/// otherwise -> cluster i is the i-th char, at `column + i * width`
/// ```
///
/// The second case is exact because a cell carrying zerowidth marks is always
/// emitted alone, so every cell in a multi-cluster span contributes exactly one
/// `char`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Span {
/// First column of the run.
@@ -54,10 +78,34 @@ pub struct Span {
/// combining marks follow its base character, so the renderer never sees
/// a base and its accent as separate glyphs.
pub text: String,
/// Columns each cluster in this run occupies: 1, or 2 for wide glyphs.
///
/// Uniform across the run by construction -- a width change ends the span.
/// This is what lets the consumer position clusters by computed origin
/// rather than by accumulated text advance.
pub width: u8,
/// How many display clusters [`text`](Self::text) holds.
///
/// Without this the consumer cannot distinguish a one-cluster span carrying
/// combining marks from an ordinary multi-character run, and would need a
/// Unicode zerowidth table to guess. The grid already knows, so it says.
pub cluster_count: u16,
/// Packed style: fg, bg, and attribute flags.
pub style: Style,
}
impl Span {
/// The decoding invariant, stated once: a span is either a single cluster
/// (which may hold several `char`s, as a keycap or an accented letter
/// does) or one cluster per `char`.
///
/// Exposed so consumers can assert it at a trust boundary rather than
/// restate it. The encoder checks it in debug builds on every frame.
pub fn counts_are_consistent(&self) -> bool {
self.cluster_count == 1 || usize::from(self.cluster_count) == self.text.chars().count()
}
}
/// Visual style of a span, as the renderer needs it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct Style {
@@ -159,6 +207,58 @@ pub fn capture(terminal: &mut crate::Terminal) -> RawFrame {
}
}
/// Copy the **entire visible viewport**, leaving damage untouched.
///
/// This exists for subscribers that arrive mid-stream: attach, reattach, and
/// the successor side of a resize. Damage only describes what changed since
/// the last capture, so a newcomer that starts from [`capture`] sees whatever
/// happened to change next -- often just the cursor's line -- painted onto a
/// blank screen. Upstream's `mark_fully_damaged` is private, so an embedder
/// cannot ask for a full frame that way.
///
/// **It must not consume damage, and that is the load-bearing property.** The
/// incumbent subscriber's next [`capture`] has to still see its rows. If this
/// called `damage()`/`reset_damage()` it would steal them, and the incumbent
/// would freeze on stale content while a newcomer's full-frame test passed.
/// The absence of those two calls below is the mechanism; `snapshot_test.rs`
/// is the proof.
///
/// **Runs under the lock; does no encoding.** Costs a full grid copy rather
/// than a damaged-rows copy, so it belongs on attach, not in the frame loop.
pub fn capture_all(terminal: &mut crate::Terminal) -> RawFrame {
let viewport = terminal.viewport();
let term = terminal.term_mut();
let columns = term.columns();
let screen_lines = term.screen_lines();
let cursor_point = term.grid().cursor.point;
let visible = term
.mode()
.contains(alacritty_terminal::term::TermMode::SHOW_CURSOR);
let grid = term.grid();
let mut rows = Vec::with_capacity(screen_lines);
for line in 0..screen_lines {
let row = &grid[Line(line as i32)];
rows.push((line, row[..Column(columns)].to_vec()));
}
let cursor = CursorFrame {
line: cursor_point.line.0.max(0) as usize,
column: cursor_point.column.0,
visible,
};
// No `damage()` and no `reset_damage()`: see the note above.
RawFrame {
rows,
cursor,
// A snapshot *is* a repaint, and marking it full also resets the
// consumer's `Encoder` hashes, so its dedup state describes the grid it
// was actually given rather than a predecessor's.
full: true,
viewport,
}
}
/// Suppresses rows whose content did not actually change.
#[derive(Default)]
pub struct Encoder {
@@ -219,9 +319,20 @@ fn hash_cells(cells: &[Cell]) -> u64 {
hasher.finish()
}
/// Group a row's cells into styled runs.
/// Group a row's cells into runs of uniform style and width.
///
/// A run continues only while style *and* width match, and a cell carrying
/// zerowidth marks is always emitted alone. Both breaks exist so the consumer
/// can compute each cluster's column as `column + i * width`; see [`Span`].
///
/// The width comparison is the only thing keeping widths uniform within a run:
/// [`Style`] deliberately excludes [`GEOMETRY_FLAGS`], so a style key cannot
/// break a run on width behind this check's back.
fn spans(cells: &[Cell]) -> Vec<Span> {
let mut spans: Vec<Span> = Vec::new();
// Whether the run in progress may still be extended. Kept here rather than
// on `Span` because it is grouping bookkeeping, not part of the wire shape.
let mut open = false;
for (column, cell) in cells.iter().enumerate() {
// A wide glyph occupies two cells: the character, then a spacer. The
// spacer carries no text of its own -- emitting its placeholder space
@@ -230,28 +341,76 @@ fn spans(cells: &[Cell]) -> Vec<Span> {
continue;
}
let style = style_of(cell);
let width = if cell.flags.contains(Flags::WIDE_CHAR) {
2
} else {
1
};
let zerowidth = cell.zerowidth();
let mut text = String::new();
text.push(cell.c);
if let Some(zerowidth) = cell.zerowidth() {
text.extend(zerowidth);
if let Some(marks) = zerowidth {
text.extend(marks);
}
// A cluster with combining marks holds more `char`s than columns, so it
// cannot share a run: it is the one case where "one char per cluster"
// stops holding.
let joinable = zerowidth.is_none();
match spans.last_mut() {
Some(last) if last.style == style => last.text.push_str(&text),
// `cluster_count` is refused rather than wrapped when it would
// overflow: the run simply ends and a new span starts at this
// column, which the consumer's rule already handles.
Some(last)
if open
&& joinable
&& last.style == style
&& last.width == width
&& last.cluster_count < u16::MAX =>
{
last.text.push_str(&text);
last.cluster_count += 1;
}
_ => spans.push(Span {
column,
text,
width,
cluster_count: 1,
style,
}),
}
open = joinable;
}
debug_assert!(
spans.iter().all(Span::counts_are_consistent),
"cluster_count must be 1 or the span's char count"
);
spans
}
/// Flags describing where a cell sits in the grid rather than how it looks.
///
/// `WRAPLINE` marks the last cell of a row that wrapped; the three wide-char
/// bits mark a two-column glyph and its spacer. Neither says anything about
/// appearance.
///
/// These are excluded from [`Style`] so the style key means one thing: visual
/// attributes. Geometry travels in [`Span::width`], which is compared on its
/// own when grouping -- if these bits stayed in the key they would break runs
/// as a side effect and leave the width comparison untestable.
///
/// Composite visual aliases (`BOLD_ITALIC`, `DIM_BOLD`, `ALL_UNDERLINES`) are
/// deliberately not masked: those are appearance.
const GEOMETRY_FLAGS: Flags = Flags::WRAPLINE
.union(Flags::WIDE_CHAR)
.union(Flags::WIDE_CHAR_SPACER)
.union(Flags::LEADING_WIDE_CHAR_SPACER);
fn style_of(cell: &Cell) -> Style {
Style {
fg: pack_color(cell.fg),
bg: pack_color(cell.bg),
flags: cell.flags.bits(),
flags: cell.flags.difference(GEOMETRY_FLAGS).bits(),
}
}

3
desktop/src-tauri/crates/buzz-terminal/src/lib.rs
View File

@@ -9,6 +9,7 @@ pub mod context;
pub mod damage;
pub mod env_fence;
pub mod fences;
pub mod lifecycle;
pub mod listener;
pub mod path;
pub mod reader;
@@ -19,6 +20,8 @@ pub mod shell;
mod context_tests;
#[cfg(test)]
mod env_fence_tests;
#[cfg(test)]
mod lifecycle_tests;
use alacritty_terminal::grid::Dimensions;
use alacritty_terminal::term::{Config, Osc52, Term};

259
desktop/src-tauri/crates/buzz-terminal/src/lifecycle.rs Normal file
View File

@@ -0,0 +1,259 @@
//! Child-process lifecycle for spawned PTY sessions.
//!
//! Closing a terminal tab must actually end the work the tab was doing. That
//! is harder than calling `kill`, for two reasons that both come from the
//! child being a *session leader* rather than an ordinary subprocess.
//!
//! **1. The child is not the only process.** `portable-pty` calls `setsid()`
//! in `pre_exec` (`unix.rs:257`), so the shell becomes a session and process
//! group leader; everything it runs — `vim`, a `make -j8` tree, a backgrounded
//! `sleep` — joins that group or a descendant of it. Signalling the shell's
//! pid alone reaches the shell. A shell that exits without forwarding the
//! signal leaves its children running, reparented to init, holding the pty
//! slave open. That is a leak that survives the window closing.
//!
//! So we signal the **process group** (`kill(-pgid)`), not the pid.
//!
//! **2. `portable-pty`'s own `kill` is not sufficient here.** `ChildKiller for
//! std::process::Child` (`lib.rs:340-373`) sends `SIGHUP` to the *pid*, waits
//! up to 4x50 ms, then falls back to `Child::kill` — which is `SIGKILL`, again
//! to the pid. Both halves are pid-scoped, so neither reaches a grandchild.
//! It is a correct API for "end this process"; ours is "end this session".
//!
//! ## The escalation
//!
//! `SIGTERM` to the group, a bounded wait for the leader, then `SIGKILL` to
//! the group **whether or not the leader went quietly** -- see `shutdown` for
//! why a polite leader does not imply an empty group.
//! `SIGTERM` first because a shell asked to terminate cleanly will flush its
//! history and let `vim` write its swap file; going straight to `SIGKILL`
//! guarantees no process ever gets that chance. The bounded wait is what makes
//! the escalation real — without it, `SIGKILL` either races the polite path
//! (making `SIGTERM` decorative) or never fires (making a signal-ignoring
//! child immortal).
//!
//! ## What this deliberately does not do
//!
//! A process that has called `setsid()` for *itself* has left our group, and
//! no group signal reaches it. `nohup`, a daemonising build tool, and
//! `tmux`-style servers all do this on purpose. We do not hunt the process
//! tree to find them: walking children to signal them is a race against a
//! moving tree — a pid read and then signalled may be a *different* process by
//! the time the signal lands, and killing a stranger's pid is a far worse bug
//! than leaking a daemon the user deliberately detached. Detaching from the
//! session is the documented way to survive one's terminal, and honouring it
//! is correct behaviour, not a gap.
use std::io;
use std::time::{Duration, Instant};
use portable_pty::Child;
/// How long the group gets to honour `SIGTERM` before `SIGKILL`.
///
/// Long enough for a shell to run its exit trap and for an editor to write a
/// swap file; short enough that closing a tab never feels stuck. Tab close is
/// not synchronous with this wait in the UI, so this is a cleanup deadline,
/// not a frame budget.
pub const TERM_GRACE: Duration = Duration::from_millis(250);
/// Poll interval while waiting for the child to exit.
///
/// Polling rather than blocking in `wait()`: a blocking wait cannot be given a
/// deadline without a second thread, and the whole point of the grace period
/// is that it expires.
const POLL_INTERVAL: Duration = Duration::from_millis(5);
/// How a session ended.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Shutdown {
/// Already gone before we signalled.
AlreadyExited,
/// Exited within [`TERM_GRACE`] of `SIGTERM`.
Terminated,
/// Ignored or outlived `SIGTERM`; the group was killed.
Killed,
}
/// Ends the session led by `child`: `SIGTERM` to its process group, a bounded
/// wait, then `SIGKILL` to the group if anything is still there.
///
/// Reaps the child before returning, so the caller cannot leave a zombie by
/// dropping the handle. Returns which arm ended it, which is what a test can
/// assert on — "did it die" is satisfied by both arms and so distinguishes
/// nothing.
#[cfg(unix)]
pub fn shutdown(child: &mut Box<dyn Child + Send + Sync>) -> io::Result<Shutdown> {
// Reap first. A child that already exited still has a pid slot until it is
// waited for, and that pid is reusable the moment it is released -- so
// signalling without checking is how a cleanup path eventually signals an
// unrelated process. Check before signalling, every time.
if child.try_wait()?.is_some() {
return Ok(Shutdown::AlreadyExited);
}
let Some(pid) = child.process_id() else {
// No pid means nothing to signal; still ensure it is reaped.
child.wait()?;
return Ok(Shutdown::AlreadyExited);
};
let pid = pid as i32;
signal_group(pid, libc::SIGTERM);
let leader_honoured_term = leader_exited_by(pid, Instant::now() + TERM_GRACE);
// Sweep the group unconditionally, *including* when the leader exited
// politely. The leader's exit is not the session's end: anything it
// backgrounded that ignores SIGTERM is still running, still in the group,
// and still holding the pty. Returning `Terminated` at that point reports
// success over a leak.
//
// Ordering with the reap is a safety requirement, not a preference. A
// process group id *is* the leader's pid, and the kernel may recycle that
// pid once the leader is reaped -- at which point `kill(-pid)` names some
// unrelated group. An exited-but-unreaped leader is a zombie, and a zombie
// is still a group member, so the id cannot be reused while we hold it.
// Signal first, reap second, and the window does not exist.
signal_group(pid, libc::SIGKILL);
// SIGKILL cannot be caught, so this terminates. It is still a `wait`
// rather than an assumption: the pid must be reaped, and the exit status
// is only available to whoever reaps it.
child.wait()?;
Ok(if leader_honoured_term {
Shutdown::Terminated
} else {
Shutdown::Killed
})
}
/// Sends `signal` to `pid`'s process group, falling back to the pid alone.
///
/// The fallback matters: `kill(-pgid)` requires the child to *be* a group
/// leader, which it is only because `portable-pty` called `setsid()`. If that
/// ever stops being true, a pid-scoped signal still ends the shell — degraded
/// (grandchildren survive) rather than a silent no-op.
///
/// Errors are deliberately not propagated. Every failure mode here means the
/// process is already gone (`ESRCH`) or was never ours to signal (`EPERM`),
/// and in both cases the following `wait` is the authority on what happened.
#[cfg(unix)]
fn signal_group(pid: i32, signal: i32) {
// SAFETY: `kill` with a negative pid targets the process group; both
// arguments are plain integers and the call has no memory effects.
let sent = unsafe { libc::kill(-pid, signal) };
if sent != 0 {
// SAFETY: as above.
unsafe { libc::kill(pid, signal) };
}
}
/// Polls until the leader has exited, or `deadline` passes. Returns whether it
/// exited in time.
///
/// Deliberately **not** `Child::try_wait`, which reaps: reaping here would
/// release the process group id before the sweep above can use it. `WNOWAIT`
/// reads the child's exit state and leaves it waitable, so the zombie stays
/// and keeps the group id reserved for us.
///
/// `waitid`, not `waitpid`, and that is a portability requirement rather than
/// taste. POSIX only defines `WNOWAIT` for `waitid`; Linux tolerates it on
/// `waitpid`, and **Darwin returns `EINVAL`**. Measured with a C probe: on
/// macOS 25.5.0, `waitpid(pid, &st, WNOHANG | WNOWAIT)` is `-1/EINVAL` for a
/// child that has plainly exited. That failure is silent in the shape this
/// function had — an error is indistinguishable from "not exited yet", so the
/// grace period could never be honoured and *every* shutdown escalated to
/// `SIGKILL`, reporting `Killed` for a child that died politely on the first
/// `SIGTERM`. The polite arm was dead code on the platform we develop on.
///
/// `waitid` reports a still-running child as success-with-`si_pid == 0`, so
/// the out-parameter must be zeroed before each call and the *pid*, not the
/// return code, is the answer.
#[cfg(unix)]
fn leader_exited_by(pid: i32, deadline: Instant) -> bool {
loop {
// SAFETY: `info` is a valid, fully-initialised out-pointer for the
// duration of the call. `WNOWAIT` leaves the child waitable, so the
// later `wait` still returns its status.
let exited = unsafe {
let mut info: libc::siginfo_t = std::mem::zeroed();
let rc = libc::waitid(
libc::P_PID,
pid as libc::id_t,
&mut info,
libc::WEXITED | libc::WNOHANG | libc::WNOWAIT,
);
rc == 0 && info.si_pid() == pid
};
if exited {
return true;
}
if Instant::now() >= deadline {
return false;
}
std::thread::sleep(POLL_INTERVAL);
}
}
/// Maximum concurrent terminal sessions.
///
/// Each session costs a pty pair (two fds), a reader thread, and a scrollback
/// grid -- at the default 10k lines x 80 cols that is megabytes of resident
/// memory per tab. The cap exists because tab creation is one keystroke and
/// nothing else bounds it: without a limit, a held-down shortcut exhausts the
/// process fd table, and the first thing to fail is not the terminal but
/// whatever *else* in Buzz next asks for a file descriptor -- the relay
/// socket, a database handle. A resource a UI can allocate in a loop needs a
/// ceiling that fails in its own subsystem.
///
/// 20 matches the abandoned `feat/terminal` branch's `MAX_LIVE_SESSIONS`,
/// kept deliberately: it is far above any plausible human tab count and far
/// below the default 256-fd soft limit, so it bounds the runaway case without
/// ever being reachable by hand.
pub const MAX_LIVE_SESSIONS: usize = 20;
/// A reader that is still consuming the PTY master, to be stopped only after
/// the session has been torn down.
///
/// This exists because the correct close order is not the obvious one, and
/// nothing in the type system otherwise prevents the wrong one. Mari's ruling
/// (`62509b91`) is that **the reader outlives child termination and reap**:
///
/// 1. mark the session closing and stop publishing to the UI;
/// 2. `SIGTERM` -> grace -> `SIGKILL` -> reap, *while output is still drained*;
/// 3. only then close the master and join the reader.
///
/// Inverting steps 2 and 3 is the bug this trait is shaped to prevent, and it
/// is not a hypothetical: a child blocked writing into a master nobody reads
/// does not die promptly even on `SIGKILL`, because the kernel completes the
/// tty teardown first. Measured with a `forkpty` probe -- **606 ms** to reap a
/// `SIGKILL`ed child against an undrained master, versus microseconds when
/// drained. Join the reader first and every tab close pays that, on the arm
/// where the user is already waiting.
pub trait DrainingReader {
/// Stops consuming and releases the reader. Called *after* reap.
fn join(self: Box<Self>);
}
/// Ends a session in the order the drain law requires, and returns how it
/// ended.
///
/// The ordering is enforced by ownership rather than by documentation: this
/// function takes the reader **by value**, so a caller cannot have joined it
/// beforehand -- a joined reader has been consumed and cannot be passed here.
/// The only way to use this API is the correct order. A comment saying "do not
/// join the reader first" is advice; a moved value is a compile error.
#[cfg(unix)]
pub fn shutdown_draining(
child: &mut Box<dyn Child + Send + Sync>,
reader: Box<dyn DrainingReader>,
) -> io::Result<Shutdown> {
// Terminate and reap with the reader still running, so the child never
// blocks in a tty write while we are waiting on it.
let outcome = shutdown(child);
// Unconditional: the reader must be released whether or not the shutdown
// reported an error, or a failed close leaks the thread and its fd.
reader.join();
outcome
}

528
desktop/src-tauri/crates/buzz-terminal/src/lifecycle_tests.rs Normal file
View File

@@ -0,0 +1,528 @@
//! Lifecycle gates: the session dies, the grandchild dies with it, and the
//! login `argv[0]` the child actually receives is the one we computed.
//!
//! Every test here drives a real PTY and a real process tree. A mock child
//! would let us assert that we *called* `kill`, which is the half we already
//! know; the property under test is what the kernel does with a process group
//! we do not fully control.
use crate::env_fence::fence_env;
use crate::lifecycle::{shutdown, shutdown_draining, DrainingReader, Shutdown, TERM_GRACE};
use crate::path::user_shell_path;
use crate::shell::{login_argv0, resolve_shell};
use portable_pty::{native_pty_system, Child, CommandBuilder, PtyPair, PtySize};
use std::sync::atomic::Ordering;
use std::time::{Duration, Instant};
/// Upper bound on any wait in this file.
///
/// Every wait here is bounded, and that is not caution -- it is the lesson
/// from a probe of an interactive child that read to EOF and hung for 300 s.
/// A PTY master does not reach EOF while any process holds the slave open, so
/// "read until the child is done" is not a terminating program. Bound the
/// read, or poll for the observable effect.
const BOUND: Duration = Duration::from_secs(10);
/// Self-destruct deadline, in seconds, for fixture processes built to ignore
/// signals.
///
/// Comfortably longer than [`BOUND`], so it can never end a process while the
/// test is still observing it -- a watchdog that fires inside the observation
/// window would make a *failing* implementation look correct. Short enough
/// that a crashed run does not leave a core spinning until reboot.
const WATCHDOG: u64 = 60;
fn open_pty() -> PtyPair {
native_pty_system()
.openpty(PtySize {
rows: 24,
cols: 80,
pixel_width: 0,
pixel_height: 0,
})
.expect("openpty")
}
/// Drains the PTY master in the background for as long as it stays open.
///
/// Not hygiene -- a correctness requirement, and the cause of a 300 s hang in
/// the first version of this file. A PTY has a small kernel buffer, and a
/// child writing into a master nobody reads blocks in `write()` once it fills.
/// A process blocked in an uninterruptible tty write does not die promptly on
/// `SIGKILL`: the signal is delivered, but the kernel finishes tearing down
/// the tty session first, so `wait()` sits there while the reap completes.
/// Measured directly with a `forkpty` C probe: with the master undrained, a
/// `SIGKILL`ed child took **606 ms** to be reaped. Every terminal in the
/// product drains its master continuously -- that is what a renderer *is* --
/// so a test that doesn't is modelling a configuration that never ships.
///
/// The consequence is worth stating for the embedder: **shutdown must not be
/// called after the reader has stopped.** Tear the session down while output
/// is still being consumed, or the grace period is spent waiting on a
/// self-inflicted stall.
fn drain(pair: &PtyPair) {
let mut reader = pair.master.try_clone_reader().expect("reader");
std::thread::spawn(move || {
use std::io::Read;
let mut buf = [0u8; 4096];
while matches!(reader.read(&mut buf), Ok(n) if n > 0) {}
});
}
/// Spawns `script` under `/bin/sh` on a real PTY, fully fenced.
fn spawn_script(pair: &PtyPair, script: &str) -> Box<dyn Child + Send + Sync> {
let shell = resolve_shell(std::env::var("SHELL").ok().as_deref());
let mut cmd = CommandBuilder::new("/bin/sh");
fence_env(&mut cmd, &user_shell_path(), &shell);
cmd.arg("-c");
cmd.arg(script);
let child = pair.slave.spawn_command(cmd).expect("spawn");
drain(pair);
child
}
/// True while `pid` exists. `kill(pid, 0)` performs the permission and
/// existence checks without delivering a signal.
fn pid_alive(pid: i32) -> bool {
// SAFETY: signal 0 delivers nothing; both arguments are integers.
unsafe { libc::kill(pid, 0) == 0 }
}
/// Polls `f` until it returns true or `BOUND` elapses; returns whether it did.
///
/// Polling for the observable state rather than sleeping a fixed duration: a
/// sleep long enough to be reliable is slow, and a sleep short enough to be
/// fast is a race that fails on a loaded machine. Both are worse than asking.
fn poll_until(mut f: impl FnMut() -> bool) -> bool {
let deadline = Instant::now() + BOUND;
while Instant::now() < deadline {
if f() {
return true;
}
std::thread::sleep(Duration::from_millis(5));
}
false
}
/// Reads a file until it is non-empty or `BOUND` elapses.
fn read_when_written(path: &std::path::Path) -> Option<String> {
let mut found = None;
poll_until(|| match std::fs::read_to_string(path) {
Ok(text) if !text.trim().is_empty() => {
found = Some(text.trim().to_owned());
true
}
_ => false,
});
found
}
/// A cooperative child exits on `SIGTERM`, so the polite arm is what ends it.
///
/// The distinction matters: `Killed` and `Terminated` both leave a dead
/// process, so asserting death alone would pass with `SIGTERM` deleted
/// entirely and the grace period reduced to a delay before `SIGKILL`.
#[test]
fn cooperative_child_exits_on_term_not_kill() {
let pair = open_pty();
let mut child = spawn_script(&pair, "sleep 30");
let pid = child.process_id().expect("pid") as i32;
let outcome = shutdown(&mut child).expect("shutdown");
assert_eq!(
outcome,
Shutdown::Terminated,
"a child that dies on SIGTERM must not have needed SIGKILL"
);
assert!(poll_until(|| !pid_alive(pid)), "child survived shutdown");
}
/// A child that ignores `SIGTERM` must still die, and the escalation must be
/// what kills it.
///
/// The fixture shape is load-bearing and my first one was vacuous. I wrote
/// `trap '' TERM; sleep 30`, which *looks* like a signal-ignoring child and
/// reported `Terminated` -- the polite arm, on a child built to defeat it.
/// The reason is that `sh` does not ignore a signal on its child's behalf: the
/// group `SIGTERM` reaches `sleep`, which has no trap and dies, and the shell
/// was blocked in `wait` on exactly that `sleep`, so it reaps it and exits
/// normally. The trap was real, the ignoring was real, and the process still
/// died on `SIGTERM` -- through a path the test wasn't looking at.
///
/// Had I not checked *which* arm fired, this would have passed for the wrong
/// reason and gone on "proving" an escalation it never exercised. The loop
/// keeps the shell itself alive: no blocking `wait` to be interrupted, so the
/// trap actually governs the shell's own fate and only `SIGKILL` can end it.
///
/// The readiness handshake closes a second, subtler version of the same
/// mistake. My loop fixture *still* reported `Terminated`, because `trap` is a
/// command the shell has to reach: a signal delivered in the interval between
/// `exec` and that line finds the default disposition and kills the shell
/// outright. Isolated with a `forkpty` probe -- identical binary, only the
/// delay before signalling changed: at 500 ms all four arms survived, at 2 ms
/// all four died with signal 15. A fixture that is only *probably* armed makes
/// this test a race whose failure mode is a false pass.
///
/// This is the arm that fails if the escalation is deleted -- and the
/// `WATCHDOG` is what makes that a *failure* rather than a hang. With
/// `SIGKILL` deleted, nothing we send can end a child that ignores `SIGTERM`,
/// so `shutdown`'s final `wait` blocks forever and the mutant is detected only
/// by the harness timing out. A test that detects a bug by never finishing is
/// indistinguishable from a broken test. The deadline converts it into a
/// bounded, reportable failure.
#[test]
fn signal_ignoring_child_is_killed_after_the_grace_period() {
let dir = tempdir("buzz-terminal-trap");
let ready = dir.join("armed");
let pair = open_pty();
// The readiness file is written *after* the trap is installed, so waiting
// on it converts "probably armed by now" into an observed fact.
let mut child = spawn_script(
&pair,
&format!(
"trap '' TERM; (sleep {WATCHDOG}; kill -9 $$) & : > {}; \
while :; do sleep 0.1; done",
ready.display()
),
);
let pid = child.process_id().expect("pid") as i32;
assert!(
poll_until(|| ready.exists()),
"child never armed its SIGTERM trap; signalling now would test a \
startup race rather than the escalation"
);
let started = Instant::now();
let outcome = shutdown(&mut child).expect("shutdown");
let elapsed = started.elapsed();
assert_eq!(
outcome,
Shutdown::Killed,
"a SIGTERM-ignoring child must be escalated to SIGKILL"
);
assert!(poll_until(|| !pid_alive(pid)), "child survived SIGKILL");
assert!(
elapsed >= TERM_GRACE,
"shutdown returned in {elapsed:?}, before the {TERM_GRACE:?} grace \
period could have elapsed -- SIGTERM was never given its chance"
);
assert!(
elapsed < BOUND,
"shutdown took {elapsed:?}; the grace period is not bounded"
);
}
/// The property the whole module exists for: a **grandchild** must not outlive
/// the session.
///
/// The fixture is deliberately hostile, and the obvious version of this test
/// proves nothing. I first wrote `sleep 30 & echo $!; wait` and mutation L2 --
/// replacing `kill(-pid)` with `kill(pid)` -- **survived it**. The reason is
/// that killing a PTY session leader makes the kernel hang up the terminal and
/// `SIGHUP` the whole foreground group, so the grandchild dies either way.
/// Isolated with a `forkpty` probe: with the master held open (no fd-closure
/// hangup) and only `SIGKILL` to the shell's pid, the grandchild was gone
/// within 200 ms while the shell itself was still unreaped. The tty hangup was
/// doing the work my group signal was being credited for.
///
/// Two properties are therefore required of the grandchild, and each closes
/// one leak in the fixture:
///
/// - it **ignores `SIGHUP`**, so the tty hangup cannot end it for us; and
/// - it **busy-loops rather than sleeping**, so it is not blocked in a call
/// that the session teardown would interrupt anyway.
///
/// With both, the probe separates cleanly: pid-only leaves the grandchild
/// alive, `kill(-pgid)` does not. That is the only shape in which this test
/// can fail for the reason it claims to test.
///
/// The `WATCHDOG` is the price of that hostility. A grandchild built to
/// survive every signal we send also survives the harness: when this test
/// legitimately fails -- as it does under mutation L1 and L2 -- it leaves a
/// process spinning a core at PPID 1, and a panicking or killed test binary
/// cannot clean up after itself. So the child carries its own deadline.
/// `SIGKILL` because that is the one signal the fixture does not trap.
#[test]
fn grandchild_does_not_outlive_the_session() {
let dir = tempdir("buzz-terminal-orphan");
let pidfile = dir.join("grandchild.pid");
let armed = dir.join("armed");
let pair = open_pty();
let mut child = spawn_script(
&pair,
&format!(
"sh -c 'trap \"\" HUP TERM; (sleep {WATCHDOG}; kill -9 $$) & \
: > {armed}; while :; do :; done' & \
echo $! > {pidfile}; wait",
armed = armed.display(),
pidfile = pidfile.display()
),
);
let shell_pid = child.process_id().expect("pid") as i32;
let grandchild: i32 = read_when_written(&pidfile)
.expect("grandchild never reported its pid")
.parse()
.expect("pid is a number");
assert!(
poll_until(|| armed.exists()),
"grandchild never armed its SIGHUP trap; the tty hangup would kill it \
regardless of how we signal, and this test could not observe the \
difference"
);
assert!(
pid_alive(grandchild),
"test setup: the grandchild must be running before we shut down"
);
assert_ne!(
grandchild, shell_pid,
"test setup: the grandchild must be a distinct process, or this \
cannot tell a group signal from a pid signal"
);
shutdown(&mut child).expect("shutdown");
assert!(
poll_until(|| !pid_alive(shell_pid)),
"the session leader survived shutdown"
);
assert!(
poll_until(|| !pid_alive(grandchild)),
"an orphaned grandchild ({grandchild}) outlived the session -- the \
signal reached the shell's pid but not its process group"
);
}
/// Shutting down an already-dead child is safe and reaps it.
///
/// Without the leading `try_wait`, this path signals a pid that the kernel may
/// already have released and reassigned.
#[test]
fn shutdown_of_an_exited_child_is_a_reap_not_a_signal() {
let pair = open_pty();
let mut child = spawn_script(&pair, "exit 0");
assert!(
poll_until(|| child.try_wait().ok().flatten().is_some()),
"child did not exit"
);
assert_eq!(
shutdown(&mut child).expect("shutdown"),
Shutdown::AlreadyExited
);
}
/// The login `argv[0]` the child **actually receives**, not the string we
/// computed.
///
/// This closes the gap flagged in `e8b567aa`: `login_argv0` and
/// `portable-pty`'s `as_command` (`cmdbuilder.rs:510-517`) were each verified
/// by reading, and agreement-by-reading is not observation.
///
/// Two things make the probe terminate where a naive one hangs. The child
/// writes `$0` to a **file** rather than the PTY -- so there is no terminal
/// echo to strip, no ANSI to parse, and no dependency on the interactive
/// shell ever reaching EOF. And the read is polled to a deadline. Credit to
/// Quinn (`fcfd69b0`), whose three failed PTY-parsing harnesses established
/// that the harness was the bug.
///
/// The explicit-prog row is the control that isolates login `argv[0]` as the
/// only variable: same shell, same PTY, same fence, no `-` prefix.
#[test]
fn default_prog_child_observes_the_login_argv0() {
let dir = tempdir("buzz-terminal-argv0");
let shell = "/bin/sh";
let default_prog = observe_argv0(&dir.join("default"), shell, true);
assert_eq!(
default_prog,
login_argv0(shell),
"the child's $0 is not the login argv0 we computed"
);
assert!(
default_prog.starts_with('-'),
"a default-prog child must be a login shell: {default_prog:?}"
);
let explicit = observe_argv0(&dir.join("explicit"), shell, false);
assert_eq!(
explicit, shell,
"control: an explicitly-invoked shell must not be given a login argv0"
);
assert_ne!(
default_prog, explicit,
"control and subject agree, so this test cannot observe the login \
prefix at all"
);
}
/// Spawns a `/bin/sh` that writes its own `$0` to `pidfile`, either as a
/// default program (login argv0 applied by portable-pty) or explicitly.
///
/// The default-prog child is an *interactive* shell with no `-c`, so it is
/// driven by writing to the PTY master -- the only way to give a login shell
/// a command is to type one.
fn observe_argv0(outfile: &std::path::Path, shell: &str, default_prog: bool) -> String {
let pair = open_pty();
let resolved = resolve_shell(Some(shell));
let mut cmd = if default_prog {
CommandBuilder::new_default_prog()
} else {
CommandBuilder::new(shell)
};
fence_env(&mut cmd, &user_shell_path(), &resolved);
if !default_prog {
cmd.arg("-c");
cmd.arg(format!("printf '%s' \"$0\" > {}", outfile.display()));
}
let mut child = pair.slave.spawn_command(cmd).expect("spawn");
drain(&pair);
drop(pair.slave);
if default_prog {
use std::io::Write;
let mut writer = pair.master.take_writer().expect("writer");
writeln!(writer, "printf '%s' \"$0\" > {}", outfile.display()).expect("write");
writer.flush().expect("flush");
// Dropping the writer closes the master's write side, which the shell
// reads as end-of-input and exits on -- no `exit` command needed, and
// nothing depends on the shell's rc files having run.
drop(writer);
}
let observed = read_when_written(outfile);
let _ = crate::lifecycle::shutdown(&mut child);
observed.unwrap_or_else(|| panic!("child never reported $0 within {BOUND:?}"))
}
/// A fresh directory for a test's artifacts, replacing any prior run's.
fn tempdir(name: &str) -> std::path::PathBuf {
let dir = std::env::temp_dir().join(name);
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).expect("create temp dir");
dir
}
/// Mari's noisy-child discriminator: the reader must still be draining
/// **while** the child is being terminated and reaped.
///
/// The property is an ordering, and an ordering cannot be observed from the
/// outcome -- both orders end with a dead child and a stopped reader. So the
/// reader records *when* it read, relative to the moment close begins, and the
/// assertion is on bytes drained after that moment.
///
/// The child is deliberately noisy: it floods the PTY continuously, so a
/// master that stops being read fills its kernel buffer within milliseconds
/// and the child blocks in `write()`. That is the state the drain law exists
/// to avoid, and a quiet child cannot produce it -- with nothing being
/// written, both orders look identical and the test proves nothing.
#[test]
fn reader_drains_through_termination_and_reap() {
let pair = open_pty();
// Flood, and keep flooding: `yes` writes until the pipe is closed or the
// process dies, so there is always more output pending than the buffer
// holds.
let mut child = spawn_noisy(&pair);
let after_close = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
let closing = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let reader = RecordingReader::spawn(&pair, after_close.clone(), closing.clone());
// Let the child get well ahead of the reader before we touch anything.
assert!(
poll_until(|| after_close.load(Ordering::Relaxed) == 0 && reader.total_bytes() > 4096),
"test setup: the child is not producing enough output to fill the pty \
buffer, so this cannot distinguish drain order"
);
closing.store(true, Ordering::Relaxed);
let started = Instant::now();
let outcome = shutdown_draining(&mut child, Box::new(reader)).expect("shutdown");
let elapsed = started.elapsed();
assert_eq!(
outcome,
Shutdown::Terminated,
"a `yes` pipeline dies on SIGTERM; SIGKILL here means it was wedged in \
a tty write against an undrained master"
);
assert!(
after_close.load(Ordering::Relaxed) > 0,
"the reader drained nothing after close began -- it was joined before \
termination, which is the ordering the drain law forbids"
);
assert!(
elapsed < TERM_GRACE,
"shutdown took {elapsed:?}, at or beyond the {TERM_GRACE:?} grace \
period: the child was blocked writing to an undrained master rather \
than exiting on SIGTERM"
);
}
/// Spawns a child that floods the PTY without pause.
fn spawn_noisy(pair: &PtyPair) -> Box<dyn Child + Send + Sync> {
let shell = resolve_shell(std::env::var("SHELL").ok().as_deref());
let mut cmd = CommandBuilder::new("/bin/sh");
fence_env(&mut cmd, &user_shell_path(), &shell);
cmd.arg("-c");
cmd.arg("while :; do echo aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa; done");
// Deliberately no `drain` here: this test owns the reader.
pair.slave.spawn_command(cmd).expect("spawn")
}
/// A [`DrainingReader`] that records how much it read after close began.
struct RecordingReader {
total: std::sync::Arc<std::sync::atomic::AtomicU64>,
handle: std::thread::JoinHandle<()>,
}
impl RecordingReader {
fn spawn(
pair: &PtyPair,
after_close: std::sync::Arc<std::sync::atomic::AtomicU64>,
closing: std::sync::Arc<std::sync::atomic::AtomicBool>,
) -> Self {
let mut reader = pair.master.try_clone_reader().expect("reader");
let total = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
let counter = total.clone();
let handle = std::thread::spawn(move || {
use std::io::Read;
let mut buf = [0u8; 4096];
while let Ok(n) = reader.read(&mut buf) {
if n == 0 {
break;
}
counter.fetch_add(n as u64, Ordering::Relaxed);
if closing.load(Ordering::Relaxed) {
after_close.fetch_add(n as u64, Ordering::Relaxed);
}
}
});
Self { total, handle }
}
fn total_bytes(&self) -> u64 {
self.total.load(Ordering::Relaxed)
}
}
impl DrainingReader for RecordingReader {
fn join(self: Box<Self>) {
// Bounded, and that is the whole point. The read loop ends when the
// master reports EOF, which only happens once the reaped child has
// released the slave -- so joining *before* termination blocks
// forever. That is precisely the forbidden ordering (mutation L4),
// and an unbounded join would "detect" it by hanging, which is
// indistinguishable from a broken test. Waiting to a deadline and
// abandoning the thread converts the hang into an assertion failure
// the harness can report.
if !poll_until(|| self.handle.is_finished()) {
return;
}
let _ = self.handle.join();
}
}

29
desktop/src-tauri/crates/buzz-terminal/src/listener.rs
View File

@@ -80,6 +80,35 @@ impl Listener {
}
}
/// Resolve an emulator action that can be answered without renderer state.
///
/// Color queries deliberately return `None`: named/indexed colors resolve
/// against the live theme, which this crate does not own.
pub fn reply(
action: Action,
columns: u16,
rows: u16,
cell_width: u16,
cell_height: u16,
) -> Option<String> {
match action {
Action::PtyWrite(text) => Some(text),
Action::SizeReply { format } => Some(format(WindowSize {
num_lines: rows,
num_cols: columns,
cell_width,
cell_height,
})),
// Palette values are renderer-owned. The transport must answer these
// only after it has a renderer palette, never invent one here.
Action::ColorReply { .. }
| Action::Title(_)
| Action::ResetTitle
| Action::Wakeup
| Action::Bell => None,
}
}
impl EventListener for Listener {
fn send_event(&self, event: Event) {
let action = match event {

28
desktop/src-tauri/crates/buzz-terminal/src/shared.rs
View File

@@ -168,6 +168,24 @@ impl SharedTerminal {
encoder.encode(raw)
}
/// Copy the whole viewport for a subscriber that arrived mid-stream.
/// Renderer plane.
///
/// Attach, reattach, and the successor side of a resize all need the
/// screen as it stands, not the next thing to change on it. Crucially this
/// leaves damage alone, so taking a snapshot for a newcomer cannot steal
/// the incumbent renderer's pending rows -- see [`damage::capture_all`].
///
/// Costs a full grid copy under the lock, so call it on attach rather than
/// per frame.
pub fn snapshot(&self, encoder: &mut Encoder) -> Frame {
let raw = {
let mut term = self.acquire(&self.renderer);
damage::capture_all(&mut term)
};
encoder.encode(raw)
}
/// Apply a coalesced resize. Renderer plane: this competes with the
/// renderer for the same lock and can hold it for milliseconds.
pub fn resize(&self, size: crate::Size) -> crate::Viewport {
@@ -181,6 +199,16 @@ impl SharedTerminal {
self.acquire(&self.renderer)
}
/// Modes the renderer/input boundary needs to report alongside frames.
pub fn input_modes(&self) -> (bool, bool) {
let term = self.acquire(&self.renderer);
let mode = term.term().mode();
(
mode.contains(alacritty_terminal::term::TermMode::BRACKETED_PASTE),
mode.contains(alacritty_terminal::term::TermMode::FOCUS_IN_OUT),
)
}
fn acquire(&self, meter: &AcquireMeter) -> MutexGuard<'_, Terminal> {
let started = Instant::now();
let guard = self.term.lock();

6
desktop/src-tauri/crates/buzz-terminal/src/shell.rs
View File

@@ -130,3 +130,9 @@ pub fn login_argv0(shell: &str) -> String {
let basename = shell.rsplit('/').next().unwrap_or(shell);
format!("-{basename}")
}
/// Resolve the command shell on Windows from `ComSpec`, falling back to cmd.
#[cfg(windows)]
pub fn resolve_shell(shell_env: Option<&str>) -> String {
shell_env.unwrap_or("cmd.exe").to_owned()
}

266
desktop/src-tauri/crates/buzz-terminal/tests/clusters.rs Normal file
View File

@@ -0,0 +1,266 @@
//! The cluster-positioning contract: what the renderer may rely on to place
//! text at the right column without consulting Unicode tables.
//!
//! The consumer's rule reads two numbers off each span and does arithmetic:
//! `cluster_count == 1` means the whole text is one cluster at `column`,
//! otherwise cluster `i` is the i-th `char` at `column + i * width`.
//!
//! These fixtures exist because that rule is not self-evidently satisfiable --
//! the two cases below require *opposite* text-splitting rules, so no encoding
//! that ships a concatenated string and a start column can be correct:
//!
//! * a regional-indicator flag is two ordinary one-column cells, so its two
//! codepoints occupy two columns and must split per codepoint;
//! * a keycap is one cell holding three codepoints, so it occupies one column
//! and must split per grapheme.
//!
//! Both are handled here by construction rather than by rule: uniform `width`
//! within a span, and a span of its own for any cluster carrying zerowidth
//! marks.
use buzz_terminal::damage::{Encoder, Span};
use buzz_terminal::fences::Fences;
use buzz_terminal::{Action, SharedTerminal, Size, Terminal};
use std::sync::mpsc::Receiver;
/// The receiver is returned rather than dropped: dropping it disconnects the
/// channel and every subsequent listener send silently fails.
fn render(input: &str) -> (Vec<Span>, Receiver<Action>) {
let size = Size {
columns: 20,
screen_lines: 2,
scrollback: 100,
};
let (term, actions) = Terminal::new(size, Fences::ALL);
let shared = SharedTerminal::new(term);
shared.feed(input.as_bytes());
let mut encoder = Encoder::new();
let frame = shared.render(&mut encoder);
let spans = frame
.rows
.into_iter()
.find(|row| row.line == 0)
.map(|row| row.spans)
.unwrap_or_default();
(spans, actions)
}
/// Apply the documented consumer rule and return `(column, cluster)` pairs,
/// dropping trailing blank padding.
///
/// This is the renderer's arithmetic, written out. Note what is *not* here: no
/// Unicode table, no zerowidth classifier, no grapheme segmentation. The
/// earlier draft of this helper carried a hand-rolled `is_zerowidth` matcher,
/// which is how we learned the encoding was under-specified -- if the fixture
/// needs a Unicode table to decode the wire, so does every real consumer.
fn placements(spans: &[Span]) -> Vec<(usize, String)> {
let mut placed = Vec::new();
for span in spans {
assert!(
span.counts_are_consistent(),
"encoder emitted an undecodable span: {span:?}"
);
let clusters: Vec<String> = if span.cluster_count == 1 {
vec![span.text.clone()]
} else {
span.text.chars().map(|c| c.to_string()).collect()
};
for (i, cluster) in clusters.into_iter().enumerate() {
if cluster != " " {
placed.push((span.column + i * span.width as usize, cluster));
}
}
}
placed
}
/// Max's case: mixed narrow and wide glyphs in one style. Every cluster must
/// land on the column the grid actually put it in.
#[test]
fn mixed_width_clusters_keep_their_columns() {
let (spans, _actions) = render("a\u{1F600}b\u{4E00}c");
assert_eq!(
placements(&spans),
vec![
(0, "a".into()),
(1, "\u{1F600}".into()),
(3, "b".into()),
(4, "\u{4E00}".into()),
(6, "c".into()),
],
"wide glyphs must advance two columns and narrow ones must not"
);
}
/// A combining mark rides with its base character and consumes no column of
/// its own, so the text that follows must not be displaced by it.
///
/// Against the previous encoding this row was a single span `"éxy"` at column
/// 0, and a consumer stepping one column per `char` placed `x` at 1 and `y`
/// at 2 -- both one column left of the truth.
#[test]
fn combining_marks_do_not_displace_following_text() {
let (spans, _actions) = render("e\u{0301}xy");
assert_eq!(
placements(&spans),
vec![(0, "e\u{0301}".into()), (1, "x".into()), (2, "y".into()),],
"a zerowidth mark must not consume a column"
);
}
/// A regional-indicator pair: two separate one-column cells. This is the case
/// that must split *per codepoint*.
#[test]
fn regional_indicator_flag_occupies_two_columns() {
let (spans, _actions) = render("\u{1F1FA}\u{1F1F8}X");
assert_eq!(
placements(&spans),
vec![
(0, "\u{1F1FA}".into()),
(1, "\u{1F1F8}".into()),
(2, "X".into()),
],
"regional indicators are one column each; X must sit at 2"
);
}
/// A keycap: one cell holding three codepoints. This is the case that must
/// split *per grapheme* -- the opposite rule from the flag above, which is why
/// the width and the cluster break both have to come from the grid.
#[test]
fn keycap_occupies_one_column() {
let (spans, _actions) = render("1\u{FE0F}\u{20E3}X");
assert_eq!(
placements(&spans),
vec![(0, "1\u{FE0F}\u{20E3}".into()), (1, "X".into()),],
"a keycap is one column; X must sit at 1"
);
}
/// Width is uniform within a span by construction. Without this a consumer
/// cannot multiply -- it would have to know each cluster's width individually,
/// which is the Unicode table this design exists to avoid.
#[test]
fn a_span_never_mixes_widths() {
let (spans, _actions) = render("ab\u{4E00}\u{4E00}cd");
for span in &spans {
let expected = span.width;
assert!(
span.width == 1 || span.width == 2,
"width must be 1 or 2, got {expected}"
);
}
let widths: Vec<u8> = spans.iter().map(|s| s.width).collect();
assert!(
widths.contains(&2),
"fixture must actually produce a wide span, got {widths:?}"
);
assert_eq!(
placements(&spans),
vec![
(0, "a".into()),
(1, "b".into()),
(2, "\u{4E00}".into()),
(4, "\u{4E00}".into()),
(6, "c".into()),
(7, "d".into()),
],
"two adjacent wide glyphs must advance two columns each"
);
}
/// `cluster_count` is what makes the wire decodable without a Unicode table,
/// so it is asserted directly here rather than only implied by placements.
///
/// The decisive pair: both spans below are width 1 with more than one `char`
/// of text, and they differ *only* in whether the count tracks the char count.
/// A consumer without that number cannot tell them apart -- which is the
/// defect Mari caught in the previous encoding.
#[test]
fn cluster_count_distinguishes_a_marked_cluster_from_a_plain_run() {
let (marked, _a) = render("e\u{0301}");
let marked = marked.first().expect("a span must be emitted");
assert_eq!(marked.text.chars().count(), 2, "base plus combining mark");
assert_eq!(marked.cluster_count, 1, "one cluster occupying one column");
// The plain run absorbs the row's blank padding, so its length is the
// viewport width rather than 2 -- what matters is that the count tracks
// the char count instead of collapsing to 1.
let (plain, _b) = render("ab");
let plain = plain.first().expect("a span must be emitted");
assert!(plain.cluster_count > 1, "a plain run is not one cluster");
assert_eq!(
usize::from(plain.cluster_count),
plain.text.chars().count(),
"one cluster per char"
);
assert_eq!(marked.width, plain.width, "both are width 1");
assert!(marked.counts_are_consistent() && plain.counts_are_consistent());
}
/// Wrapping marks the last cell of the row with `WRAPLINE` (upstream
/// `term/mod.rs:968`). That bit records where the text happened to wrap, not
/// how the text looks, so it must not reach the style key: if it did, the last
/// column of every wrapped row would split off into a span of its own -- an
/// extra wire record per wrapped line, and span boundaries that move when the
/// window is resized.
///
/// Quinn found this by reading `cell.rs:21` while checking the `WIDE_CHAR`
/// mask; this fixture is the proof that was missing from the source read.
#[test]
fn wrapping_does_not_split_a_uniform_run() {
let size = Size {
columns: 5,
screen_lines: 3,
scrollback: 100,
};
let (term, _actions) = Terminal::new(size, Fences::ALL);
let shared = SharedTerminal::new(term);
// Six narrow cells in one style: five fill row 0 and set WRAPLINE on the
// last of them, the sixth lands on row 1.
shared.feed(b"abcdef");
let mut encoder = Encoder::new();
let frame = shared.render(&mut encoder);
let first = frame
.rows
.iter()
.find(|row| row.line == 0)
.expect("wrapped row must be present");
let texts: Vec<&str> = first.spans.iter().map(|s| s.text.as_str()).collect();
assert_eq!(
texts,
vec!["abcde"],
"a wrapped row of one style is one span; WRAPLINE must not break it"
);
}
/// A wide glyph at the last usable column wraps to the next row rather than
/// straddling the edge. The contract must hold on the wrapped row too.
#[test]
fn leading_wide_glyph_after_wrap_is_positioned_from_column_zero() {
let size = Size {
columns: 5,
screen_lines: 3,
scrollback: 100,
};
let (term, _actions) = Terminal::new(size, Fences::ALL);
let shared = SharedTerminal::new(term);
// Four narrow cells fill 0..=3, leaving one column: the wide glyph cannot
// fit and moves to the next row.
shared.feed("abcd\u{4E00}".as_bytes());
let mut encoder = Encoder::new();
let frame = shared.render(&mut encoder);
let second = frame
.rows
.iter()
.find(|row| row.line == 1)
.expect("wrapped row must be present");
assert_eq!(
placements(&second.spans),
vec![(0, "\u{4E00}".into())],
"a wrapped wide glyph starts at column 0 of the next row"
);
}

187
desktop/src-tauri/crates/buzz-terminal/tests/snapshot.rs Normal file
View File

@@ -0,0 +1,187 @@
//! The attach contract: what a subscriber that arrives mid-stream is given,
//! and what taking it must not cost the subscriber already there.
//!
//! `render()` reports damage -- what changed since someone last looked. That
//! is the right thing for a steady-state renderer and the wrong thing for a
//! newcomer, who needs the screen as it stands. `snapshot()` supplies that,
//! and the delicate part is that it must do so *without* consuming damage:
//! two subscribers share one terminal, and damage is a single shared cursor.
use buzz_terminal::damage::Encoder;
use buzz_terminal::fences::Fences;
use buzz_terminal::{Action, SharedTerminal, Size, Terminal};
use std::sync::mpsc::Receiver;
/// The receiver is returned rather than dropped: dropping it disconnects the
/// channel and every subsequent listener send silently fails.
fn terminal(columns: usize, screen_lines: usize) -> (SharedTerminal, Receiver<Action>) {
let size = Size {
columns,
screen_lines,
scrollback: 100,
};
let (term, actions) = Terminal::new(size, Fences::ALL);
(SharedTerminal::new(term), actions)
}
/// Collect the non-blank text of a frame's rows, for comparing what a
/// subscriber can actually see.
fn visible_text(frame: &buzz_terminal::damage::Frame) -> Vec<String> {
frame
.rows
.iter()
.map(|row| {
row.spans
.iter()
.map(|span| span.text.as_str())
.collect::<String>()
.trim_end()
.to_string()
})
.filter(|line| !line.is_empty())
.collect()
}
/// The reason `snapshot` exists. A subscriber that attaches mid-stream and
/// starts from `render()` is handed only what changes next -- with a quiet
/// terminal that is the cursor's line alone, so the scrollback-visible screen
/// never arrives.
#[test]
fn a_late_render_shows_only_the_next_change_but_a_snapshot_shows_the_screen() {
let (shared, _actions) = terminal(20, 4);
shared.feed(b"first\r\nsecond\r\nthird");
// The incumbent consumes the damage from that output.
let mut incumbent = Encoder::new();
let seen = visible_text(&shared.render(&mut incumbent));
assert_eq!(seen, vec!["first", "second", "third"]);
// A newcomer rendering now sees essentially nothing: damage is spent.
let mut latecomer = Encoder::new();
let by_render = visible_text(&shared.render(&mut latecomer));
assert!(
!by_render.contains(&"first".to_string()),
"a late render cannot show scrollback it never saw damaged, got {by_render:?}"
);
// The same newcomer snapshotting sees the whole viewport.
let mut attaching = Encoder::new();
let by_snapshot = shared.snapshot(&mut attaching);
assert_eq!(
visible_text(&by_snapshot),
vec!["first", "second", "third"],
"a snapshot must carry the visible viewport"
);
assert!(by_snapshot.full, "a snapshot is a repaint");
}
/// **The law: `snapshot()` must not consume damage.**
///
/// Two subscribers share one terminal and damage is one shared cursor, so a
/// snapshot taken for an attaching subscriber must leave the incumbent's
/// pending rows intact. A naive implementation that calls `damage()` passes a
/// full-frame test while freezing every other subscriber -- the newcomer looks
/// perfect and the incumbent silently stops updating.
///
/// The interleaving is the point: write, snapshot, *then* let the incumbent
/// render. But the interleaving alone is not enough to discriminate, and the
/// reason is this module's own rule 2 -- `Term::damage()` marks the cursor
/// line on every call. So an incumbent owed only the line it is sitting on
/// gets that line back even when its damage was stolen, and a naive snapshot
/// passes.
///
/// The owed row therefore has to be somewhere the cursor is *not*. Here row 0
/// is rewritten and the cursor is parked on row 3, so a theft leaves the
/// incumbent holding a blank cursor line and nothing else.
#[test]
fn a_snapshot_does_not_steal_the_incumbents_damage() {
let (shared, _actions) = terminal(20, 4);
// An established renderer, caught up to a quiet terminal. The initial
// content is shorter than its replacement so the rewrite below covers it
// completely and no tail of it survives.
let mut incumbent = Encoder::new();
shared.feed(b"old");
let _ = shared.render(&mut incumbent);
// Rewrite row 0, then park the cursor on row 3. The incumbent is now owed
// row 0, which is not the row the cursor will re-damage for free.
shared.feed(b"\x1b[1;1HAFTER\x1b[4;1H");
// A second subscriber attaches and snapshots first.
let mut attaching = Encoder::new();
let attached = shared.snapshot(&mut attaching);
assert_eq!(
visible_text(&attached),
vec!["AFTER"],
"the newcomer sees the whole screen"
);
// The incumbent must still be delivered row 0.
let follow_up = shared.render(&mut incumbent);
assert!(
follow_up.rows.iter().any(|row| row.line == 0),
"snapshot consumed the incumbent's damage: row 0 was never delivered, \
got rows {:?}",
follow_up.rows.iter().map(|r| r.line).collect::<Vec<_>>()
);
assert!(
visible_text(&follow_up).contains(&"AFTER".to_string()),
"the incumbent must still see the row written before the snapshot, got {:?}",
visible_text(&follow_up)
);
}
/// A snapshot stamps the geometry it was captured under and resets the
/// consumer's dedup state, so an encoder reused across a resize cannot carry
/// hashes describing rows of a different width.
#[test]
fn a_snapshot_realigns_a_reused_encoders_dedup_state() {
let (shared, _actions) = terminal(20, 4);
shared.feed(b"wide enough line");
let mut encoder = Encoder::new();
let before = shared.snapshot(&mut encoder);
assert_eq!(before.viewport.columns, 20);
let resized = shared.resize(Size {
columns: 10,
screen_lines: 4,
scrollback: 100,
});
assert_eq!(resized.columns, 10);
// Same encoder, new geometry: every row must be re-sent, not suppressed
// as unchanged against hashes taken at the old width.
let after = shared.snapshot(&mut encoder);
assert_eq!(
after.viewport.columns, 10,
"the capture-time grid is stamped"
);
assert!(after.full, "a snapshot is a repaint");
assert!(
!after.rows.is_empty(),
"stale hashes must not suppress rows after a resize"
);
}
/// Two consecutive snapshots with no output between them still both carry the
/// screen. A snapshot is not a one-shot: reattach may happen repeatedly, and
/// nothing about the first may disarm the second.
#[test]
fn snapshots_are_repeatable() {
let (shared, _actions) = terminal(20, 4);
shared.feed(b"persistent");
let mut first = Encoder::new();
let mut second = Encoder::new();
assert_eq!(
visible_text(&shared.snapshot(&mut first)),
vec!["persistent"]
);
assert_eq!(
visible_text(&shared.snapshot(&mut second)),
vec!["persistent"],
"a second subscriber attaching later must see the same screen"
);
}

12
desktop/src-tauri/src/lib.rs
View File

@@ -32,6 +32,9 @@ mod reset;
mod secret_store;
mod shutdown;
mod templates;
mod terminal_runtime;
#[cfg_attr(not(test), allow(dead_code))]
mod terminal_transport;
#[cfg(target_os = "macos")]
mod tray_menu;
mod util;
@@ -368,6 +371,7 @@ pub fn run() {
.manage(BuilderlabSession::default())
.manage(BuilderlabLogin::default())
.manage(commands::pairing::PairingHandle::new())
.manage(terminal_runtime::TerminalSessions::default())
.setup(move |app| {
let app_handle = app.handle().clone();
#[cfg(target_os = "macos")]
@@ -663,6 +667,14 @@ pub fn run() {
Ok(())
})
.invoke_handler(tauri::generate_handler![
terminal_runtime::terminal_attach,
terminal_runtime::terminal_detach,
terminal_runtime::terminal_close,
terminal_runtime::terminal_input,
terminal_runtime::terminal_resize,
terminal_runtime::terminal_ack,
terminal_runtime::terminal_viewport_ready,
terminal_runtime::terminal_focus,
take_pending_community_deep_link,
acknowledge_pending_community_deep_link,
start_builderlab_login,

4
desktop/src-tauri/src/shutdown.rs
View File

@@ -19,6 +19,8 @@ pub(crate) fn shut_down_app(app: &tauri::AppHandle, shutdown_done: &std::sync::a
.store(true, Ordering::SeqCst);
if !shutdown_done.swap(true, Ordering::SeqCst) {
prevent_sleep::release(&app.state::<AppState>().prevent_sleep);
app.state::<crate::terminal_runtime::TerminalSessions>()
.shutdown_all();
if let Err(error) = shutdown_managed_agents(app) {
eprintln!("buzz-desktop: failed to stop managed agents: {error}");
}
@@ -40,6 +42,8 @@ pub(crate) fn install_signal_handler(
.shutdown_started
.store(true, Ordering::SeqCst);
if !shutdown_done.swap(true, Ordering::SeqCst) {
app.state::<crate::terminal_runtime::TerminalSessions>()
.shutdown_all();
let _ = shutdown_managed_agents(&app);
#[cfg(feature = "mesh-llm")]
shutdown_mesh_runtime(&app);

856
desktop/src-tauri/src/terminal_runtime.rs Normal file
View File

@@ -0,0 +1,856 @@
//! Rust-owned PTY sessions and the typed Tauri transport for Buzz Substrate.
use std::io::{Read, Write};
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;
use buzz_terminal::context::{context_vars, GuiContext};
use buzz_terminal::damage::{Frame, Style};
use buzz_terminal::{Fences, SharedTerminal, Size, Terminal, Viewport};
use portable_pty::{native_pty_system, CommandBuilder, MasterPty, PtySize};
use serde::{Deserialize, Serialize};
use tauri::ipc::Channel;
use uuid::Uuid;
use crate::terminal_transport::{FramePublisher, OfferError, Publication, SubscriptionId};
const MAX_LIVE_SESSIONS: usize = 20;
const MAX_INPUT_BYTES: usize = 1024 * 1024;
type Result<T> = std::result::Result<T, String>;
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct AttachRequest {
/// Present when a renderer remounts onto an existing PTY-backed tab.
session_id: Option<String>,
channel_id: String,
channel_name: String,
thread_id: Option<String>,
npub: String,
relay_url: String,
columns: u16,
rows: u16,
pixel_width: u16,
pixel_height: u16,
}
#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub(crate) struct WireViewport {
generation: u64,
columns: usize,
screen_lines: usize,
}
impl From<Viewport> for WireViewport {
fn from(value: Viewport) -> Self {
Self {
generation: value.generation,
columns: value.columns,
screen_lines: value.screen_lines,
}
}
}
impl From<WireViewport> for Viewport {
fn from(value: WireViewport) -> Self {
Self {
generation: value.generation,
columns: value.columns,
screen_lines: value.screen_lines,
}
}
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct AttachResponse {
session_id: String,
subscription_id: String,
viewport: WireViewport,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct WireStyle {
fg: u32,
bg: u32,
flags: u16,
}
impl From<Style> for WireStyle {
fn from(value: Style) -> Self {
Self {
fg: value.fg,
bg: value.bg,
flags: value.flags,
}
}
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct WireCluster {
column: usize,
text: String,
width: u8,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct WireSpan {
style: WireStyle,
clusters: Vec<WireCluster>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct WireRow {
line: usize,
spans: Vec<WireSpan>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct WireCursor {
line: usize,
column: usize,
visible: bool,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct FrameMessage {
subscription_id: String,
sequence: u64,
rows: Vec<WireRow>,
cursor: WireCursor,
full: bool,
viewport: WireViewport,
bracketed_paste: bool,
focus_reporting: bool,
}
#[derive(Debug, Serialize)]
#[serde(tag = "type", content = "payload", rename_all = "camelCase")]
pub(crate) enum TerminalMessage {
Frame(FrameMessage),
Title(String),
ResetTitle,
Bell,
Exit,
}
fn wire_publication(publication: Publication) -> Result<FrameMessage> {
let frame = publication.frame;
let rows = frame
.rows
.into_iter()
.map(|row| {
let spans = row
.spans
.into_iter()
.map(|span| {
if !span.counts_are_consistent() {
return Err(
"terminal engine emitted an inconsistent cluster count".to_string()
);
}
let clusters = if span.cluster_count == 1 {
vec![WireCluster {
column: span.column,
text: span.text,
width: span.width,
}]
} else {
span.text
.chars()
.enumerate()
.map(|(index, ch)| WireCluster {
column: span.column + index * usize::from(span.width),
text: ch.to_string(),
width: span.width,
})
.collect()
};
Ok(WireSpan {
style: span.style.into(),
clusters,
})
})
.collect::<Result<Vec<_>>>()?;
Ok(WireRow {
line: row.line,
spans,
})
})
.collect::<Result<Vec<_>>>()?;
Ok(FrameMessage {
subscription_id: publication.subscription_id.to_string(),
sequence: publication.sequence,
rows,
cursor: WireCursor {
line: frame.cursor.line,
column: frame.cursor.column,
visible: frame.cursor.visible,
},
full: frame.full,
viewport: frame.viewport.into(),
bracketed_paste: false,
focus_reporting: false,
})
}
struct ReaderThread(Option<JoinHandle<()>>);
impl buzz_terminal::lifecycle::DrainingReader for ReaderThread {
fn join(mut self: Box<Self>) {
if let Some(handle) = self.0.take() {
let _ = handle.join();
}
}
}
struct Session {
id: Uuid,
terminal: Arc<SharedTerminal>,
master: Option<Box<dyn MasterPty + Send>>,
writer: Arc<Mutex<Box<dyn Write + Send>>>,
pty_size: Arc<Mutex<PtySize>>,
child: Box<dyn portable_pty::Child + Send + Sync>,
reader: Option<Box<dyn buzz_terminal::lifecycle::DrainingReader + Send>>,
publisher: Arc<Mutex<FramePublisher>>,
channel: Arc<Mutex<Option<Channel<TerminalMessage>>>>,
}
impl Session {
fn publish(&self, publication: Publication) {
let subscription = publication.subscription_id;
let sent = wire_publication(publication).and_then(|message| {
let (bracketed_paste, focus_reporting) = self.terminal.input_modes();
let mut message = message;
message.bracketed_paste = bracketed_paste;
message.focus_reporting = focus_reporting;
self.channel
.lock()
.map_err(|e| e.to_string())?
.as_ref()
.ok_or_else(|| "terminal renderer detached".to_string())?
.send(TerminalMessage::Frame(message))
.map_err(|e| e.to_string())
});
if sent.is_err() {
if let Ok(mut channel) = self.channel.lock() {
*channel = None;
}
if let Ok(mut publisher) = self.publisher.lock() {
publisher.fault(subscription);
}
}
}
fn shutdown(mut self) {
if let Ok(mut channel) = self.channel.lock() {
*channel = None;
}
// The slave closes on child reap; the reader continues draining until then.
if let Some(reader) = self.reader.take() {
#[cfg(unix)]
{
let _ = buzz_terminal::lifecycle::shutdown_draining(&mut self.child, reader);
}
#[cfg(not(unix))]
{
let _ = self.child.kill();
let _ = self.child.wait();
reader.join();
}
}
self.master.take();
}
}
#[derive(Default)]
pub(crate) struct TerminalSessions(Mutex<Vec<Session>>);
impl TerminalSessions {
fn with_session<T>(&self, id: &str, f: impl FnOnce(&mut Session) -> Result<T>) -> Result<T> {
let id = Uuid::parse_str(id).map_err(|_| "invalid terminal session id".to_string())?;
let mut sessions = self.0.lock().map_err(|e| e.to_string())?;
let session = sessions
.iter_mut()
.find(|session| session.id == id)
.ok_or_else(|| "terminal session not found".to_string())?;
f(session)
}
pub(crate) fn shutdown_all(&self) {
let sessions = self
.0
.lock()
.map(|mut sessions| std::mem::take(&mut *sessions))
.unwrap_or_default();
for session in sessions {
session.shutdown();
}
}
}
fn size(columns: u16, rows: u16) -> Result<Size> {
if columns == 0 || rows == 0 {
return Err("terminal dimensions must be non-zero".to_string());
}
Ok(Size {
columns: usize::from(columns),
screen_lines: usize::from(rows),
..Size::default()
})
}
fn pty_size(columns: u16, rows: u16, pixel_width: u16, pixel_height: u16) -> PtySize {
PtySize {
rows,
cols: columns,
pixel_width,
pixel_height,
}
}
fn offer_capture(
publisher: &Mutex<FramePublisher>,
frame: Frame,
snapshot: impl FnOnce() -> Frame,
) -> Option<Publication> {
let offered = publisher.lock().ok()?.offer(frame);
match offered {
Ok(publication) => publication,
Err(OfferError::PendingFrameMustBeSnapshot) => {
publisher.lock().ok()?.offer(snapshot()).ok().flatten()
}
}
}
#[tauri::command]
pub(crate) fn terminal_attach(
request: AttachRequest,
on_frame: Channel<TerminalMessage>,
state: tauri::State<'_, TerminalSessions>,
) -> Result<AttachResponse> {
let terminal_size = size(request.columns, request.rows)?;
let mut sessions = state.0.lock().map_err(|e| e.to_string())?;
if let Some(existing_id) = request.session_id.as_deref() {
let existing_id =
Uuid::parse_str(existing_id).map_err(|_| "invalid terminal session id".to_string())?;
let session = sessions
.iter_mut()
.find(|session| session.id == existing_id)
.ok_or_else(|| "terminal session not found".to_string())?;
let subscription = SubscriptionId::new();
let mut encoder = buzz_terminal::damage::Encoder::new();
let snapshot = session.terminal.snapshot(&mut encoder);
let viewport = snapshot.viewport;
*session.channel.lock().map_err(|e| e.to_string())? = Some(on_frame);
let bootstrap = session
.publisher
.lock()
.map_err(|e| e.to_string())?
.attach(subscription, snapshot)
.map_err(|_| "terminal snapshot rejected".to_string())?;
session.publish(bootstrap);
return Ok(AttachResponse {
session_id: existing_id.to_string(),
subscription_id: subscription.to_string(),
viewport: viewport.into(),
});
}
if sessions.len() >= MAX_LIVE_SESSIONS {
return Err("too many live terminal sessions".to_string());
}
let id = Uuid::new_v4();
let pair = native_pty_system()
.openpty(pty_size(
request.columns,
request.rows,
request.pixel_width,
request.pixel_height,
))
.map_err(|e| e.to_string())?;
let shell = {
#[cfg(unix)]
{
buzz_terminal::shell::resolve_shell(std::env::var("SHELL").ok().as_deref())
}
#[cfg(windows)]
{
buzz_terminal::shell::resolve_shell(std::env::var("ComSpec").ok().as_deref())
}
};
let mut command = CommandBuilder::new_default_prog();
buzz_terminal::env_fence::fence_env(
&mut command,
&buzz_terminal::path::user_shell_path(),
&shell,
);
let context = GuiContext {
channel_id: request.channel_id,
channel_name: request.channel_name,
thread_id: request.thread_id,
npub: request.npub,
relay_url: request.relay_url,
session_id: id.to_string(),
};
for (key, value) in context_vars(&context) {
command.env(key, value);
}
let child = pair
.slave
.spawn_command(command)
.map_err(|e| e.to_string())?;
drop(pair.slave);
let mut reader = pair.master.try_clone_reader().map_err(|e| e.to_string())?;
let writer = Arc::new(Mutex::new(
pair.master.take_writer().map_err(|e| e.to_string())?,
));
let (terminal, actions) = Terminal::new(terminal_size, Fences::default());
let terminal = Arc::new(SharedTerminal::new(terminal));
let publisher = Arc::new(Mutex::new(FramePublisher::new(terminal.lock().viewport())));
let channel = Arc::new(Mutex::new(Some(on_frame)));
let current_pty_size = Arc::new(Mutex::new(pty_size(
request.columns,
request.rows,
request.pixel_width,
request.pixel_height,
)));
// Emulator replies (DSR/size) must go back through the PTY even when no
// renderer is attached. The action thread ends when the terminal drops.
let action_writer = Arc::clone(&writer);
let action_size = Arc::clone(&current_pty_size);
let action_channel = Arc::clone(&channel);
std::thread::spawn(move || {
while let Ok(action) = actions.recv() {
let presentation = match &action {
buzz_terminal::Action::Title(title) => Some(TerminalMessage::Title(title.clone())),
buzz_terminal::Action::ResetTitle => Some(TerminalMessage::ResetTitle),
buzz_terminal::Action::Bell => Some(TerminalMessage::Bell),
_ => None,
};
if let Some(message) = presentation {
let _ = action_channel
.lock()
.ok()
.and_then(|channel| channel.as_ref()?.send(message).ok());
continue;
}
let size = action_size.lock().map(|size| *size).unwrap_or_default();
if let Some(reply) = buzz_terminal::listener::reply(
action,
size.cols,
size.rows,
size.pixel_width,
size.pixel_height,
) {
if action_writer.lock().map_or(true, |mut writer| {
writer.write_all(reply.as_bytes()).is_err()
}) {
break;
}
}
}
});
let reader_terminal = Arc::clone(&terminal);
let reader_publisher = Arc::clone(&publisher);
let reader_channel = Arc::clone(&channel);
let reader_handle = std::thread::spawn(move || {
let mut buffer = [0u8; 16 * 1024];
let mut encoder = buzz_terminal::damage::Encoder::new();
let mut snapshot_encoder = buzz_terminal::damage::Encoder::new();
loop {
let count = match reader.read(&mut buffer) {
Ok(0) | Err(_) => break,
Ok(count) => count,
};
reader_terminal.feed(&buffer[..count]);
let needs_snapshot = reader_publisher
.lock()
.map(|publisher| publisher.requires_snapshot())
.unwrap_or(false);
let frame = if needs_snapshot {
reader_terminal.snapshot(&mut snapshot_encoder)
} else {
reader_terminal.render(&mut encoder)
};
if frame.is_empty() {
continue;
}
let publication = offer_capture(&reader_publisher, frame, || {
reader_terminal.snapshot(&mut snapshot_encoder)
});
if let Some(publication) = publication {
let subscription = publication.subscription_id;
let result = wire_publication(publication).and_then(|mut message| {
let (bracketed_paste, focus_reporting) = reader_terminal.input_modes();
message.bracketed_paste = bracketed_paste;
message.focus_reporting = focus_reporting;
reader_channel
.lock()
.map_err(|e| e.to_string())?
.as_ref()
.ok_or_else(|| "detached".to_string())?
.send(TerminalMessage::Frame(message))
.map_err(|e| e.to_string())
});
if result.is_err() {
if let Ok(mut publisher) = reader_publisher.lock() {
publisher.fault(subscription);
}
}
}
}
let _ = reader_channel
.lock()
.ok()
.and_then(|channel| channel.as_ref()?.send(TerminalMessage::Exit).ok());
});
let subscription = SubscriptionId::new();
let mut snapshot_encoder = buzz_terminal::damage::Encoder::new();
let viewport = terminal.lock().viewport();
let bootstrap = publisher
.lock()
.map_err(|e| e.to_string())?
.attach(subscription, terminal.snapshot(&mut snapshot_encoder))
.map_err(|_| "terminal snapshot rejected".to_string())?;
let session = Session {
id,
terminal,
master: Some(pair.master),
writer,
pty_size: current_pty_size,
child,
reader: Some(Box::new(ReaderThread(Some(reader_handle)))),
publisher,
channel,
};
session.publish(bootstrap);
sessions.push(session);
Ok(AttachResponse {
session_id: id.to_string(),
subscription_id: subscription.to_string(),
viewport: viewport.into(),
})
}
#[tauri::command]
pub(crate) fn terminal_detach(
session_id: String,
subscription_id: String,
state: tauri::State<'_, TerminalSessions>,
) -> Result<()> {
state.with_session(&session_id, |session| {
let id = SubscriptionId::parse(&subscription_id)?;
let detached = session
.publisher
.lock()
.map_err(|e| e.to_string())?
.fault(id);
if detached {
*session.channel.lock().map_err(|e| e.to_string())? = None;
}
Ok(())
})
}
#[tauri::command]
pub(crate) fn terminal_close(
session_id: String,
state: tauri::State<'_, TerminalSessions>,
) -> Result<()> {
let id = Uuid::parse_str(&session_id).map_err(|_| "invalid terminal session id".to_string())?;
let session = {
let mut sessions = state.0.lock().map_err(|e| e.to_string())?;
let index = sessions
.iter()
.position(|session| session.id == id)
.ok_or_else(|| "terminal session not found".to_string())?;
sessions.remove(index)
};
session.shutdown();
Ok(())
}
#[tauri::command]
pub(crate) fn terminal_input(
session_id: String,
data: String,
state: tauri::State<'_, TerminalSessions>,
) -> Result<()> {
if data.len() > MAX_INPUT_BYTES {
return Err("terminal input exceeds 1 MiB".to_string());
}
state.with_session(&session_id, |session| {
session
.writer
.lock()
.map_err(|e| e.to_string())?
.write_all(data.as_bytes())
.map_err(|e| e.to_string())
})
}
#[tauri::command]
pub(crate) fn terminal_resize(
session_id: String,
columns: u16,
rows: u16,
pixel_width: u16,
pixel_height: u16,
state: tauri::State<'_, TerminalSessions>,
) -> Result<WireViewport> {
let terminal_size = size(columns, rows)?;
state.with_session(&session_id, |session| {
let new_pty_size = pty_size(columns, rows, pixel_width, pixel_height);
session
.master
.as_ref()
.ok_or_else(|| "terminal is closing".to_string())?
.resize(new_pty_size)
.map_err(|e| e.to_string())?;
*session.pty_size.lock().map_err(|e| e.to_string())? = new_pty_size;
let viewport = session.terminal.resize(terminal_size);
let publication = {
let mut publisher = session.publisher.lock().map_err(|e| e.to_string())?;
publisher.resize_applied(viewport);
let mut encoder = buzz_terminal::damage::Encoder::new();
publisher
.offer(session.terminal.snapshot(&mut encoder))
.map_err(|_| "terminal resize snapshot rejected".to_string())?
};
if let Some(publication) = publication {
session.publish(publication);
}
Ok(viewport.into())
})
}
#[tauri::command]
pub(crate) fn terminal_ack(
session_id: String,
subscription_id: String,
sequence: u64,
state: tauri::State<'_, TerminalSessions>,
) -> Result<()> {
state.with_session(&session_id, |session| {
let id = SubscriptionId::parse(&subscription_id)?;
let publication = session
.publisher
.lock()
.map_err(|e| e.to_string())?
.acknowledge(id, sequence);
if let Some(publication) = publication {
session.publish(publication);
}
Ok(())
})
}
#[tauri::command]
pub(crate) fn terminal_viewport_ready(
session_id: String,
subscription_id: String,
viewport: WireViewport,
state: tauri::State<'_, TerminalSessions>,
) -> Result<()> {
state.with_session(&session_id, |session| {
let id = SubscriptionId::parse(&subscription_id)?;
let publication = session
.publisher
.lock()
.map_err(|e| e.to_string())?
.viewport_ready(id, viewport.into());
if let Some(publication) = publication {
session.publish(publication);
}
Ok(())
})
}
#[tauri::command]
pub(crate) fn terminal_focus(
session_id: String,
focused: bool,
state: tauri::State<'_, TerminalSessions>,
) -> Result<()> {
state.with_session(&session_id, |session| {
let (_, enabled) = session.terminal.input_modes();
if enabled {
session
.writer
.lock()
.map_err(|e| e.to_string())?
.write_all(if focused { b"\x1b[I" } else { b"\x1b[O" })
.map_err(|e| e.to_string())?;
}
Ok(())
})
}
#[cfg(test)]
mod tests {
use super::*;
use buzz_terminal::damage::{CursorFrame, RowFrame, Span};
fn publication(spans: Vec<Span>) -> Publication {
Publication {
subscription_id: SubscriptionId::new(),
sequence: 7,
frame: buzz_terminal::damage::Frame {
rows: vec![RowFrame { line: 3, spans }],
cursor: CursorFrame {
line: 1,
column: 2,
visible: true,
},
full: true,
viewport: Viewport {
generation: 4,
columns: 80,
screen_lines: 24,
},
},
}
}
fn style() -> Style {
Style {
fg: 1,
bg: 2,
flags: 3,
}
}
fn marker_frame(marker: usize, full: bool) -> Frame {
Frame {
rows: vec![RowFrame {
line: marker,
spans: Vec::new(),
}],
cursor: CursorFrame {
line: 0,
column: 0,
visible: true,
},
full,
viewport: Viewport {
generation: 0,
columns: 80,
screen_lines: 24,
},
}
}
fn assert_post_snapshot_capture_survives_attach(mut publisher: FramePublisher) {
assert!(!publisher.requires_snapshot());
let bootstrap = marker_frame(1, true);
let post_snapshot_incremental = marker_frame(42, false);
let subscription = SubscriptionId::new();
publisher.attach(subscription, bootstrap).unwrap();
let publisher = Mutex::new(publisher);
assert_eq!(
offer_capture(&publisher, post_snapshot_incremental, || marker_frame(
42, true
)),
None
);
let successor = publisher
.lock()
.unwrap()
.acknowledge(subscription, 1)
.expect("post-snapshot PTY output was lost");
assert_eq!(successor.frame.rows[0].line, 42);
assert!(successor.frame.full);
}
#[test]
fn initial_attach_retains_output_captured_after_its_bootstrap_snapshot() {
let viewport = marker_frame(0, true).viewport;
let publisher = FramePublisher::new(viewport);
assert_post_snapshot_capture_survives_attach(publisher);
}
#[test]
fn reattach_retains_output_captured_after_its_bootstrap_snapshot() {
let viewport = marker_frame(0, true).viewport;
let mut publisher = FramePublisher::new(viewport);
let old = SubscriptionId::new();
publisher.attach(old, marker_frame(0, true)).unwrap();
assert!(publisher.acknowledge(old, 1).is_none());
assert_post_snapshot_capture_survives_attach(publisher);
}
#[test]
fn mapper_expands_ascii_runs_without_unicode_classification() {
let message = wire_publication(publication(vec![Span {
column: 4,
text: "abc".into(),
width: 1,
cluster_count: 3,
style: style(),
}]))
.unwrap();
let clusters = &message.rows[0].spans[0].clusters;
assert_eq!(
clusters
.iter()
.map(|cluster| (cluster.column, cluster.text.as_str()))
.collect::<Vec<_>>(),
vec![(4, "a"), (5, "b"), (6, "c")]
);
}
#[test]
fn mapper_keeps_a_multi_char_cluster_atomic() {
let message = wire_publication(publication(vec![Span {
column: 9,
text: "1\u{fe0f}\u{20e3}".into(),
width: 2,
cluster_count: 1,
style: style(),
}]))
.unwrap();
let clusters = &message.rows[0].spans[0].clusters;
assert_eq!(clusters.len(), 1);
assert_eq!(clusters[0].column, 9);
assert_eq!(clusters[0].text, "1\u{fe0f}\u{20e3}");
assert_eq!(clusters[0].width, 2);
}
#[test]
fn mapper_rejects_an_inconsistent_engine_span() {
let result = wire_publication(publication(vec![Span {
column: 0,
text: "ab".into(),
width: 1,
cluster_count: 3,
style: style(),
}]));
assert!(result.is_err());
}
#[test]
fn dimensions_reject_zero_and_preserve_scrollback_default() {
assert!(size(0, 24).is_err());
assert!(size(80, 0).is_err());
let size = size(100, 40).unwrap();
assert_eq!(
(size.columns, size.screen_lines, size.scrollback),
(100, 40, 10_000)
);
}
}

375
desktop/src-tauri/src/terminal_transport.rs Normal file
View File

@@ -0,0 +1,375 @@
//! Credit and viewport ordering for one renderer subscription.
//!
//! Tauri channels are ordered, but they do not expose consumer credit and
//! channel delivery is unordered with respect to invoke responses. This state
//! machine therefore permits one frame on the wire, retains at most one newer
//! complete snapshot, and independently gates a resized viewport until the
//! renderer confirms that it knows the applied [`Viewport`].
use buzz_terminal::{damage::Frame, Viewport};
use std::fmt;
use uuid::Uuid;
/// A renderer attachment. Messages from an attachment that has been replaced
/// cannot release the new attachment's credit.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub(crate) struct SubscriptionId(Uuid);
impl SubscriptionId {
pub(crate) fn new() -> Self {
Self(Uuid::new_v4())
}
pub(crate) fn parse(value: &str) -> Result<Self, String> {
Uuid::parse_str(value)
.map(Self)
.map_err(|_| "invalid terminal subscription id".to_string())
}
}
impl fmt::Display for SubscriptionId {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
self.0.fmt(formatter)
}
}
/// A frame carrying the identity and sequence that its ACK must repeat.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct Publication {
pub(crate) subscription_id: SubscriptionId,
pub(crate) sequence: u64,
pub(crate) frame: Frame,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum OfferError {
/// Replacing pending incremental damage would lose changes that existed in
/// the older pending frame. Only a complete snapshot may be replaceable.
PendingFrameMustBeSnapshot,
}
struct Subscription {
id: SubscriptionId,
next_sequence: u64,
in_flight: Option<u64>,
pending: Option<Frame>,
viewport_ready: bool,
}
/// Publication state for a terminal session.
///
/// PTY ownership is deliberately elsewhere. Faulting or detaching this state
/// stops renderer publication only; it must never stop the PTY reader.
pub(crate) struct FramePublisher {
applied: Viewport,
subscription: Option<Subscription>,
}
impl FramePublisher {
pub(crate) fn new(applied: Viewport) -> Self {
Self {
applied,
subscription: None,
}
}
pub(crate) fn applied_viewport(&self) -> Viewport {
self.applied
}
/// Attach with a complete snapshot of the current viewport. Attachment is
/// a fresh bootstrap: no credit or readiness from the old subscription is
/// inherited.
pub(crate) fn attach(
&mut self,
id: SubscriptionId,
snapshot: Frame,
) -> Result<Publication, OfferError> {
self.require_current_snapshot(&snapshot)?;
let mut subscription = Subscription {
id,
next_sequence: 1,
in_flight: None,
pending: None,
viewport_ready: true,
};
let publication = Self::publish(&mut subscription, snapshot);
self.subscription = Some(subscription);
Ok(publication)
}
/// Whether the next capture must be complete so it can safely replace the
/// single pending value. Callers use this to select the engine's snapshot
/// path before taking the terminal lock.
pub(crate) fn requires_snapshot(&self) -> bool {
self.subscription.as_ref().is_some_and(|subscription| {
subscription.in_flight.is_some() || !subscription.viewport_ready
})
}
/// Offer the newest capture. Frames for an overtaken viewport are inert.
/// While credit/readiness is held, the newest complete snapshot replaces
/// the prior one and pending storage remains exactly one frame.
pub(crate) fn offer(&mut self, frame: Frame) -> Result<Option<Publication>, OfferError> {
if frame.viewport != self.applied {
return Ok(None);
}
let Some(subscription) = &mut self.subscription else {
return Ok(None);
};
if subscription.in_flight.is_some() || !subscription.viewport_ready {
if !frame.full {
return Err(OfferError::PendingFrameMustBeSnapshot);
}
subscription.pending = Some(frame);
return Ok(None);
}
Ok(Some(Self::publish(subscription, frame)))
}
/// Apply the viewport returned by the engine resize call. A real change
/// invalidates pending state and closes the readiness gate. Same-size
/// resizes are inert because the engine returns the unchanged viewport.
pub(crate) fn resize_applied(&mut self, viewport: Viewport) {
if viewport == self.applied {
return;
}
self.applied = viewport;
if let Some(subscription) = &mut self.subscription {
subscription.viewport_ready = false;
subscription.pending = None;
}
}
/// Release ordinary frame credit only for the exact live subscription and
/// exact frame in flight.
pub(crate) fn acknowledge(&mut self, id: SubscriptionId, sequence: u64) -> Option<Publication> {
let subscription = self.subscription.as_mut()?;
if subscription.id != id || subscription.in_flight != Some(sequence) {
return None;
}
subscription.in_flight = None;
Self::flush(subscription)
}
/// Renderer confirmation that the resize invoke result has been installed.
/// Equality against the one atomic viewport value is the entire policy:
/// old subscription IDs and superseded viewport values are inert.
pub(crate) fn viewport_ready(
&mut self,
id: SubscriptionId,
viewport: Viewport,
) -> Option<Publication> {
let subscription = self.subscription.as_mut()?;
if subscription.id != id || viewport != self.applied {
return None;
}
subscription.viewport_ready = true;
Self::flush(subscription)
}
/// Drop publication state after channel close, invoke rejection, renderer
/// teardown, or explicit detach. The caller owns safe Buzz focus recovery.
pub(crate) fn fault(&mut self, id: SubscriptionId) -> bool {
if self
.subscription
.as_ref()
.is_some_and(|subscription| subscription.id == id)
{
self.subscription = None;
true
} else {
false
}
}
fn require_current_snapshot(&self, frame: &Frame) -> Result<(), OfferError> {
if frame.viewport == self.applied && frame.full {
Ok(())
} else {
Err(OfferError::PendingFrameMustBeSnapshot)
}
}
fn flush(subscription: &mut Subscription) -> Option<Publication> {
if subscription.in_flight.is_some() || !subscription.viewport_ready {
return None;
}
let frame = subscription.pending.take()?;
Some(Self::publish(subscription, frame))
}
fn publish(subscription: &mut Subscription, frame: Frame) -> Publication {
debug_assert!(subscription.in_flight.is_none());
let sequence = subscription.next_sequence;
subscription.next_sequence = subscription
.next_sequence
.checked_add(1)
.expect("terminal publication sequence exhausted");
subscription.in_flight = Some(sequence);
Publication {
subscription_id: subscription.id,
sequence,
frame,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use buzz_terminal::damage::{CursorFrame, RowFrame};
fn viewport(generation: u64, columns: usize, screen_lines: usize) -> Viewport {
Viewport {
generation,
columns,
screen_lines,
}
}
fn frame(viewport: Viewport, marker: usize, full: bool) -> Frame {
Frame {
rows: vec![RowFrame {
line: marker,
spans: Vec::new(),
}],
cursor: CursorFrame {
line: 0,
column: 0,
visible: true,
},
full,
viewport,
}
}
#[test]
fn attach_bootstraps_then_ack_releases_only_latest_snapshot() {
let v0 = viewport(0, 80, 24);
let mut publisher = FramePublisher::new(v0);
let id = SubscriptionId::new();
let bootstrap = publisher.attach(id, frame(v0, 0, true)).unwrap();
assert_eq!(bootstrap.sequence, 1);
assert!(publisher.requires_snapshot());
assert_eq!(publisher.offer(frame(v0, 1, true)).unwrap(), None);
assert_eq!(publisher.offer(frame(v0, 2, true)).unwrap(), None);
let successor = publisher.acknowledge(id, 1).unwrap();
assert_eq!(successor.sequence, 2);
assert_eq!(successor.frame.rows[0].line, 2);
}
#[test]
fn pending_incremental_frame_is_rejected_instead_of_losing_damage() {
let v0 = viewport(0, 80, 24);
let mut publisher = FramePublisher::new(v0);
let id = SubscriptionId::new();
publisher.attach(id, frame(v0, 0, true)).unwrap();
assert_eq!(
publisher.offer(frame(v0, 1, false)),
Err(OfferError::PendingFrameMustBeSnapshot)
);
}
#[test]
fn late_old_frame_is_acked_but_never_crosses_viewports() {
let v0 = viewport(0, 80, 24);
let v1 = viewport(1, 100, 30);
let mut publisher = FramePublisher::new(v0);
let id = SubscriptionId::new();
publisher.attach(id, frame(v0, 0, true)).unwrap();
publisher.resize_applied(v1);
assert_eq!(publisher.offer(frame(v0, 7, true)).unwrap(), None);
assert_eq!(publisher.offer(frame(v1, 8, true)).unwrap(), None);
assert_eq!(publisher.acknowledge(id, 1), None);
let current = publisher.viewport_ready(id, v1).unwrap();
assert_eq!(current.frame.viewport, v1);
assert_eq!(current.frame.rows[0].line, 8);
}
#[test]
fn early_new_frame_is_held_despite_real_capture_pressure() {
let v0 = viewport(0, 80, 24);
let v1 = viewport(1, 120, 40);
let mut publisher = FramePublisher::new(v0);
let id = SubscriptionId::new();
publisher.attach(id, frame(v0, 0, true)).unwrap();
publisher.acknowledge(id, 1);
publisher.resize_applied(v1);
// A real non-empty capture occurred. Zero publication before readiness
// therefore cannot pass vacuously because no output happened.
let captured = frame(v1, 9, true);
assert!(!captured.rows.is_empty());
assert_eq!(publisher.offer(captured).unwrap(), None);
let publication = publisher.viewport_ready(id, v1).unwrap();
assert_eq!(publication.frame.rows[0].line, 9);
assert_eq!(publication.frame.viewport, v1);
}
#[test]
fn ack_and_readiness_may_arrive_in_either_order() {
let v0 = viewport(0, 80, 24);
let v1 = viewport(1, 100, 30);
let mut publisher = FramePublisher::new(v0);
let id = SubscriptionId::new();
publisher.attach(id, frame(v0, 0, true)).unwrap();
publisher.resize_applied(v1);
publisher.offer(frame(v1, 1, true)).unwrap();
assert_eq!(publisher.viewport_ready(id, v1), None);
assert!(publisher.acknowledge(id, 1).is_some());
}
#[test]
fn superseded_readiness_and_same_size_resize_are_inert() {
let v0 = viewport(0, 80, 24);
let v1 = viewport(1, 100, 30);
let v2 = viewport(2, 120, 40);
let mut publisher = FramePublisher::new(v0);
let id = SubscriptionId::new();
publisher.attach(id, frame(v0, 0, true)).unwrap();
publisher.resize_applied(v1);
publisher.resize_applied(v2);
publisher.resize_applied(v2);
assert_eq!(publisher.applied_viewport(), v2);
publisher.offer(frame(v2, 2, true)).unwrap();
publisher.acknowledge(id, 1);
assert_eq!(publisher.viewport_ready(id, v1), None);
assert!(publisher.viewport_ready(id, v2).is_some());
}
#[test]
fn old_subscription_messages_cannot_release_reattached_subscription() {
let v0 = viewport(0, 80, 24);
let mut publisher = FramePublisher::new(v0);
let old = SubscriptionId::new();
publisher.attach(old, frame(v0, 0, true)).unwrap();
assert!(publisher.fault(old));
let new = SubscriptionId::new();
publisher.attach(new, frame(v0, 1, true)).unwrap();
publisher.offer(frame(v0, 2, true)).unwrap();
assert_eq!(publisher.acknowledge(old, 1), None);
assert_eq!(publisher.viewport_ready(old, v0), None);
let successor = publisher.acknowledge(new, 1).unwrap();
assert_eq!(successor.frame.rows[0].line, 2);
}
#[test]
fn stale_fault_cannot_detach_current_subscription() {
let v0 = viewport(0, 80, 24);
let mut publisher = FramePublisher::new(v0);
let old = SubscriptionId::new();
let new = SubscriptionId::new();
publisher.attach(old, frame(v0, 0, true)).unwrap();
publisher.attach(new, frame(v0, 1, true)).unwrap();
assert!(!publisher.fault(old));
assert!(publisher.fault(new));
}
}