mirador 1.22.2

//! Child lifecycle and bounded stdio workers for one external panel.

use std::collections::VecDeque;
use std::io::{self, BufRead, BufReader, Write};
use std::process::{Child, Command, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{self, Receiver, RecvTimeoutError, SyncSender, TrySendError};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};

use crate::config::PluginConfig;

use super::{
    DEFAULT_REFRESH, HostMessage, InputPolicy, MAX_BINDING_ACTION_BYTES, MAX_BINDING_KEY_BYTES,
    MAX_BINDINGS, MAX_COLOR_BYTES, MAX_COUNTER_BYTES, MAX_ERROR_BYTES, MAX_FRAME_LINES,
    MAX_FRAME_SPANS, MAX_INPUT_KEYS, MAX_MESSAGE_BYTES, MAX_RETAINED_FRAME_TEXT_BYTES,
    MAX_TITLE_BYTES, MAX_WATCH_TEXT_BYTES, PROTOCOL_VERSION, PluginMessage, WireBinding, WireFrame,
    WireLine, binding_mentions_host_key, host_owned_chord, is_ctrl_c_chord, wire_color,
};

const MAX_STDERR_BYTES: usize = 8 * 1024;
const EVENT_QUEUE: usize = 256;
const WATCH_QUEUE: usize = 64;
const SHUTDOWN_GRACE: Duration = Duration::from_millis(300);
const OUTPUT_CLOSE_GRACE: Duration = Duration::from_millis(100);
const STARTUP_TIMEOUT: Duration = Duration::from_secs(5);
// Pipe closure and lifecycle commands wake the supervisor immediately. This is
// only the fallback for a descendant that inherited stdout, so an idle plugin
// does not need the old permanent 50 Hz process poll.
const PROCESS_STATUS_POLL: Duration = Duration::from_secs(1);
const SUPERVISOR_JOIN_GRACE: Duration = Duration::from_millis(750);
const SUPERVISOR_ABORT_GRACE: Duration = Duration::from_millis(250);
const MESSAGE_RATE_WINDOW: Duration = Duration::from_secs(1);
const MAX_MESSAGES_PER_WINDOW: usize = 120;
const MAX_STDOUT_BYTES_PER_WINDOW: usize = 16 * 1024 * 1024;
const MAX_STDERR_BYTES_PER_WINDOW: usize = 64 * 1024;

#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) enum Phase {
    Starting,
    Running,
    Exited(String),
    Failed(String),
    Stopping,
}

pub(super) struct Shared {
    pub(super) generation: u64,
    /// Number of valid frame messages accepted, including stale revisions.
    /// Input acknowledgement is about receipt, not revision arithmetic.
    pub(super) frame_sequence: u64,
    pub(super) ready: bool,
    pub(super) phase: Phase,
    pub(super) title: Option<String>,
    pub(super) refresh: Duration,
    pub(super) frame: Option<Arc<WireFrame>>,
    pub(super) notice: Option<String>,
    pub(super) stderr: String,
    pub(super) watch: VecDeque<String>,
}

impl Shared {
    pub(super) fn starting() -> Self {
        Self {
            generation: 0,
            frame_sequence: 0,
            ready: false,
            phase: Phase::Starting,
            title: None,
            refresh: DEFAULT_REFRESH,
            frame: None,
            notice: None,
            stderr: String::new(),
            watch: VecDeque::new(),
        }
    }

    pub(super) fn changed(&mut self) {
        self.generation = self.generation.wrapping_add(1);
    }

    pub(super) fn fail(&mut self, message: impl Into<String>) {
        // Stdio workers and the supervisor can discover the same failure in a
        // different order. Preserve the first terminal explanation instead of
        // replacing a useful protocol error with the broken pipe caused by
        // terminating that process.
        if matches!(
            self.phase,
            Phase::Exited(_) | Phase::Failed(_) | Phase::Stopping
        ) {
            return;
        }
        self.phase = Phase::Failed(message.into());
        self.changed();
    }
}

enum SupervisorCommand {
    Shutdown,
    Abort,
    OutputClosed,
}

pub(super) struct Runtime {
    events: Option<SyncSender<HostMessage>>,
    /// Set once the host has asked the child to stop. The writer reads it
    /// rather than the queue, so `shutdown` does not wait for room behind
    /// input the child has not read yet — see [`writer_loop`].
    stopping: Arc<AtomicBool>,
    /// Set while a `tick` is in the queue, so there is never more than one
    /// — see [`Runtime::send_tick`]. The writer clears it as it takes the
    /// tick out.
    tick_pending: Arc<AtomicBool>,
    supervisor: mpsc::Sender<SupervisorCommand>,
    supervisor_thread: Option<JoinHandle<()>>,
}

impl Runtime {
    pub(super) fn send(&self, message: HostMessage) -> Result<(), TrySendError<HostMessage>> {
        debug_assert!(
            !matches!(message, HostMessage::Tick),
            "a tick goes through `send_tick`, which keeps it to one in the queue"
        );
        let Some(events) = self.events.as_ref() else {
            return Err(TrySendError::Disconnected(message));
        };
        events.try_send(message)
    }

    /// Queue a `tick` unless one is already waiting.
    ///
    /// A tick is a cue to redraw, not a clock: it carries nothing, so two
    /// waiting say no more than one. Queued on every poll like input, they
    /// piled up behind a plugin that had stopped reading as soon as the pipe
    /// was full — 4,096 of them on macOS, two minutes and a quarter at
    /// 33 ms on a focused panel capturing input, seventeen minutes at the
    /// dashboard's default 250 ms otherwise — and took all 256 places in
    /// 256 polls more, after which every key was dropped under a notice
    /// that blamed input. Now the queue holds at most one, however many
    /// polls pass, and the rest of it is left for input. Ticks already in
    /// the pipe are beyond reach and still arrive.
    ///
    /// The writer clears the flag *before* writing the tick it takes out,
    /// so a tick raised while that write blocks is queued, not folded into
    /// one already gone. A tick that finds no room clears it again, so the
    /// next poll tries afresh rather than finding one "waiting" that
    /// never went in. `&mut self` keeps this to one caller at a time, which
    /// is what makes clearing it on failure safe.
    pub(super) fn send_tick(&mut self) {
        if self.tick_pending.swap(true, Ordering::AcqRel) {
            return;
        }
        let queued = self
            .events
            .as_ref()
            .is_some_and(|events| events.try_send(HostMessage::Tick).is_ok());
        if !queued {
            self.tick_pending.store(false, Ordering::Release);
        }
    }

    pub(super) fn abort(&self) {
        let _ = self.supervisor.send(SupervisorCommand::Abort);
    }

    /// Ask the child to stop, then remain alive long enough for the supervisor
    /// to enforce the grace-then-kill contract. The wait itself is bounded so
    /// a broken OS process primitive cannot trap Mirador on exit.
    pub(super) fn shutdown(&mut self) -> bool {
        if let Some(events) = self.events.take() {
            // Not a message in the queue: a full queue refused it, and the
            // child that had fallen 256 messages behind — the one most in
            // need of the word — was killed without it. The flag
            // reaches the writer however full the queue is, and dropping
            // the only sender wakes a writer waiting on an empty one.
            self.stopping.store(true, Ordering::Release);
            drop(events);
            let _ = self.supervisor.send(SupervisorCommand::Shutdown);
        }

        let Some(thread) = self.supervisor_thread.take() else {
            return true;
        };
        if wait_until_finished(&thread, SUPERVISOR_JOIN_GRACE) {
            return thread.join().is_ok();
        }

        // The supervisor normally killed the process after 300 ms. Abort is a
        // second, independent nudge before the bounded host wait runs out.
        let _ = self.supervisor.send(SupervisorCommand::Abort);
        if wait_until_finished(&thread, SUPERVISOR_ABORT_GRACE) {
            thread.join().is_ok()
        } else {
            false
        }
    }

    #[cfg(test)]
    pub(super) fn stub() -> (Self, Receiver<HostMessage>) {
        let (events, received) = mpsc::sync_channel(EVENT_QUEUE);
        let (supervisor, commands) = mpsc::channel();
        let supervisor_thread = thread::spawn(move || {
            while let Ok(command) = commands.recv() {
                if matches!(
                    command,
                    SupervisorCommand::Shutdown | SupervisorCommand::Abort
                ) {
                    return;
                }
            }
        });
        (
            Self {
                events: Some(events),
                stopping: Arc::new(AtomicBool::new(false)),
                tick_pending: Arc::new(AtomicBool::new(false)),
                supervisor,
                supervisor_thread: Some(supervisor_thread),
            },
            received,
        )
    }
}

impl Drop for Runtime {
    fn drop(&mut self) {
        let _ = self.shutdown();
    }
}

fn wait_until_finished(thread: &JoinHandle<()>, timeout: Duration) -> bool {
    let deadline = Instant::now() + timeout;
    while !thread.is_finished() && Instant::now() < deadline {
        thread::sleep(Duration::from_millis(5));
    }
    thread.is_finished()
}

pub(super) fn spawn_process(
    spec: &PluginConfig,
    shared: &Arc<Mutex<Shared>>,
) -> io::Result<Runtime> {
    let mut command = Command::new(&spec.command[0]);
    command
        .args(&spec.command[1..])
        .env("MIRADOR_PLUGIN_PROTOCOL", PROTOCOL_VERSION.to_string())
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    let mut child = command.spawn()?;
    let startup_deadline = Instant::now() + STARTUP_TIMEOUT;
    let stdin = child
        .stdin
        .take()
        .ok_or_else(|| io::Error::other("missing stdin"))?;
    let stdout = child
        .stdout
        .take()
        .ok_or_else(|| io::Error::other("missing stdout"))?;
    let stderr = child
        .stderr
        .take()
        .ok_or_else(|| io::Error::other("missing stderr"))?;

    let (event_tx, event_rx) = mpsc::sync_channel(EVENT_QUEUE);
    let (supervisor_tx, supervisor_rx) = mpsc::channel();
    let stopping = Arc::new(AtomicBool::new(false));
    let tick_pending = Arc::new(AtomicBool::new(false));

    let writer_shared = Arc::clone(shared);
    let writer_supervisor = supervisor_tx.clone();
    let writer_stopping = Arc::clone(&stopping);
    let writer_tick_pending = Arc::clone(&tick_pending);
    thread::spawn(move || {
        writer_loop(
            stdin,
            &event_rx,
            &writer_stopping,
            &writer_tick_pending,
            &writer_shared,
            &writer_supervisor,
        );
    });

    let reader_shared = Arc::clone(shared);
    let reader_supervisor = supervisor_tx.clone();
    thread::spawn(move || reader_loop(stdout, &reader_shared, &reader_supervisor));

    let stderr_shared = Arc::clone(shared);
    let stderr_supervisor = supervisor_tx.clone();
    thread::spawn(move || stderr_loop(stderr, &stderr_shared, &stderr_supervisor));

    let supervisor_shared = Arc::clone(shared);
    let supervisor_thread = thread::spawn(move || {
        supervisor_loop(
            &mut child,
            &supervisor_rx,
            &supervisor_shared,
            startup_deadline,
        );
    });

    let cwd = std::env::current_dir()
        .map(|path| path.to_string_lossy().into_owned())
        .unwrap_or_default();
    let config = serde_json::to_value(&spec.config).unwrap_or(serde_json::Value::Null);
    let hello = HostMessage::Hello {
        protocol: PROTOCOL_VERSION,
        host_version: env!("CARGO_PKG_VERSION"),
        plugin: spec.id.clone(),
        config,
        cwd,
    };
    let runtime = Runtime {
        events: Some(event_tx),
        stopping,
        tick_pending,
        supervisor: supervisor_tx,
        supervisor_thread: Some(supervisor_thread),
    };
    runtime
        .send(hello)
        .map_err(|error| io::Error::other(format!("sending hello: {error}")))?;
    Ok(runtime)
}

/// Write the queued messages to the child, in order, until the host stops.
///
/// Once `stopping` is set the next thing this thread writes is `shutdown`,
/// and whatever is still in the queue is skipped. What it has already
/// written is beyond reach: a child that has stopped reading holds this
/// thread in a write only once the pipe's own buffer is full, so if it reads
/// again inside its grace it reads that buffer first and then `shutdown`,
/// not the queue's backlog as well, which it would be killed halfway
/// through. The one exception is `hello`, which is never skipped: the
/// protocol says `shutdown` may *follow* it during startup, and a plugin
/// should not be told to stop by a host that never said hello.
///
/// A `tick` taken out of the queue clears `tick_pending` before it is
/// written, which is what lets [`Runtime::send_tick`] queue the next one.
fn writer_loop(
    mut stdin: impl Write,
    events: &Receiver<HostMessage>,
    stopping: &AtomicBool,
    tick_pending: &AtomicBool,
    shared: &Arc<Mutex<Shared>>,
    supervisor: &mpsc::Sender<SupervisorCommand>,
) {
    let mut write = |event: &HostMessage| match write_host_message(&mut stdin, event) {
        Ok(()) => true,
        Err(error) => {
            shared
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .fail(format!("writing to plugin: {error}"));
            let _ = supervisor.send(SupervisorCommand::Abort);
            false
        }
    };
    for event in events {
        let event =
            if stopping.load(Ordering::Acquire) && !matches!(event, HostMessage::Hello { .. }) {
                HostMessage::Shutdown
            } else {
                event
            };
        if matches!(event, HostMessage::Tick) {
            // Before the write, not after it. A plugin slow to read holds
            // this thread inside the write, and the refreshes meanwhile
            // must queue one tick behind it, not find the flag still set
            // and fold themselves into this one, already out of the queue.
            tick_pending.store(false, Ordering::Release);
        }
        if !write(&event) || matches!(event, HostMessage::Shutdown) {
            return;
        }
    }
    // The queue is empty and its sender gone. If it went because the host
    // is stopping, the child is still owed its `shutdown`.
    if stopping.load(Ordering::Acquire) {
        write(&HostMessage::Shutdown);
    }
}

fn write_host_message(mut target: impl Write, event: &HostMessage) -> io::Result<()> {
    let encoded = serde_json::to_vec(event).map_err(io::Error::other)?;
    if encoded.len().saturating_add(1) > MAX_MESSAGE_BYTES {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            format!("host message exceeds {MAX_MESSAGE_BYTES} bytes"),
        ));
    }
    target.write_all(&encoded)?;
    target.write_all(b"\n")?;
    target.flush()
}

#[derive(Default)]
struct MessageRate {
    recent: VecDeque<(Instant, usize)>,
    bytes: usize,
}

impl MessageRate {
    fn accepts(&mut self, now: Instant, bytes: usize) -> bool {
        while self
            .recent
            .front()
            .is_some_and(|(then, _)| now.saturating_duration_since(*then) >= MESSAGE_RATE_WINDOW)
        {
            if let Some((_, expired)) = self.recent.pop_front() {
                self.bytes = self.bytes.saturating_sub(expired);
            }
        }
        if self.recent.len() >= MAX_MESSAGES_PER_WINDOW
            || self.bytes.saturating_add(bytes) > MAX_STDOUT_BYTES_PER_WINDOW
        {
            return false;
        }
        self.recent.push_back((now, bytes));
        self.bytes += bytes;
        true
    }
}

fn reader_loop(
    stdout: impl io::Read,
    shared: &Arc<Mutex<Shared>>,
    supervisor: &mpsc::Sender<SupervisorCommand>,
) {
    let mut reader = BufReader::new(stdout);
    let mut rate = MessageRate::default();
    loop {
        let mut bytes = Vec::new();
        match read_limited_line(&mut reader, &mut bytes) {
            Ok(0) => {
                let _ = supervisor.send(SupervisorCommand::OutputClosed);
                break;
            }
            Ok(_) => {}
            Err(error) => {
                shared
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .fail(format!("reading plugin output: {error}"));
                let _ = supervisor.send(SupervisorCommand::Abort);
                break;
            }
        }
        if !rate.accepts(Instant::now(), bytes.len()) {
            shared
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .fail(format!(
                    "plugin exceeded {MAX_MESSAGES_PER_WINDOW} messages or {MAX_STDOUT_BYTES_PER_WINDOW} output bytes in one second"
                ));
            let _ = supervisor.send(SupervisorCommand::Abort);
            break;
        }
        let message = match decode_line(&bytes) {
            Ok(message) => message,
            Err(error) => {
                shared
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .fail(format!("invalid plugin message: {error}"));
                let _ = supervisor.send(SupervisorCommand::Abort);
                break;
            }
        };
        if !apply_message(message, shared) {
            let _ = supervisor.send(SupervisorCommand::Abort);
            break;
        }
    }
}

/// One line of plugin output as a message: the line ending stripped — a
/// `\r\n` from a plugin written on Windows as much as a `\n` — and what is
/// left decoded.
pub(super) fn decode_line(line: &[u8]) -> serde_json::Result<PluginMessage> {
    let mut end = line.len();
    while end > 0 && matches!(line[end - 1], b'\n' | b'\r') {
        end -= 1;
    }
    serde_json::from_slice(&line[..end])
}

pub(super) fn read_limited_line(
    reader: &mut impl BufRead,
    target: &mut Vec<u8>,
) -> io::Result<usize> {
    let mut limited = io::Read::take(
        reader,
        u64::try_from(MAX_MESSAGE_BYTES + 1).unwrap_or(u64::MAX),
    );
    let read = limited.read_until(b'\n', target)?;
    if read > MAX_MESSAGE_BYTES || (read == MAX_MESSAGE_BYTES && !target.ends_with(b"\n")) {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            format!("message exceeds {MAX_MESSAGE_BYTES} bytes"),
        ));
    }
    if read > 0 && !target.ends_with(b"\n") {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            "plugin message is not terminated by a newline",
        ));
    }
    Ok(read)
}

pub(super) fn apply_message(message: PluginMessage, shared: &Arc<Mutex<Shared>>) -> bool {
    match message {
        PluginMessage::Ready {
            protocol,
            title,
            refresh_ms,
        } => apply_ready_message(protocol, title, refresh_ms, shared),
        PluginMessage::Frame {
            revision,
            title,
            counter,
            lines,
            bindings,
            input,
            cursor,
        } => apply_frame_message(
            WireFrame {
                revision,
                title,
                counter,
                lines,
                bindings,
                input,
                cursor,
            },
            shared,
        ),
        PluginMessage::Error { message, fatal } => apply_error_message(message, fatal, shared),
        PluginMessage::Watch { text } => apply_watch_message(text, shared),
    }
}

fn apply_ready_message(
    protocol: u16,
    title: Option<String>,
    refresh_ms: Option<u64>,
    shared: &Arc<Mutex<Shared>>,
) -> bool {
    if protocol != PROTOCOL_VERSION {
        return fail_shared(
            shared,
            format!("protocol {protocol} is incompatible with host protocol {PROTOCOL_VERSION}"),
        );
    }
    if let Some(title) = title.as_deref()
        && let Err(error) = validate_text(title, "ready title", MAX_TITLE_BYTES)
    {
        return fail_shared(shared, error);
    }
    let mut shared = shared
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    apply_ready(title, refresh_ms, &mut shared)
}

fn apply_frame_message(frame: WireFrame, shared: &Arc<Mutex<Shared>>) -> bool {
    if let Err(error) = validate_frame(
        frame.title.as_deref(),
        frame.counter.as_deref(),
        &frame.lines,
        &frame.bindings,
        &frame.input,
    ) {
        return fail_shared(shared, error);
    }

    let mut shared = shared
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    if stopping(&shared) {
        return true;
    }
    if !shared.ready {
        shared.fail("plugin sent a frame before `ready`");
        return false;
    }
    if !matches!(shared.phase, Phase::Running) {
        return false;
    }
    apply_frame(frame, &mut shared)
}

fn apply_error_message(message: String, fatal: bool, shared: &Arc<Mutex<Shared>>) -> bool {
    if let Err(error) = validate_text(&message, "error message", MAX_ERROR_BYTES) {
        return fail_shared(shared, error);
    }
    let mut shared = shared
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    if stopping(&shared) {
        return true;
    }
    if !shared.ready {
        shared.fail("plugin sent an error before `ready`");
        return false;
    }
    if !matches!(shared.phase, Phase::Running) {
        return false;
    }
    if fatal {
        shared.fail(message);
        return false;
    }
    shared.notice = Some(message);
    shared.changed();
    true
}

fn apply_watch_message(text: String, shared: &Arc<Mutex<Shared>>) -> bool {
    if text.trim().is_empty()
        || text.len() > MAX_WATCH_TEXT_BYTES
        || text.chars().any(char::is_control)
    {
        return fail_shared(
            shared,
            format!(
                "plugin watch text must be one non-empty line of at most \
                 {MAX_WATCH_TEXT_BYTES} UTF-8 bytes"
            ),
        );
    }
    let mut shared = shared
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    if stopping(&shared) {
        return true;
    }
    if !shared.ready {
        shared.fail("plugin sent a watch event before `ready`");
        return false;
    }
    if !matches!(shared.phase, Phase::Running) {
        return false;
    }
    if shared.watch.len() == WATCH_QUEUE {
        shared.watch.pop_front();
        shared.notice = Some("plugin watch queue overflowed; oldest event dropped".into());
    }
    shared.watch.push_back(text);
    shared.changed();
    true
}

/// Whether the host has sent `shutdown` and is waiting out the grace, in
/// which case a well-formed message is dropped rather than refused.
///
/// The supervisor moves to `Stopping` the moment it is told, but the plugin
/// learns of `shutdown` only once the writer has flushed it, so a frame it was
/// already sending arrives after the host has stopped listening. A refusal is
/// an abort, and an abort kills at once: the 300 ms the protocol promises for
/// cleanup went to any plugin that happened to be drawing. A malformed
/// message is still refused — that is checked before this is asked — because
/// the grace is for a plugin that is stopping, not for one that is broken.
/// Order is not checked: a `ready` that crossed `shutdown` in flight is
/// dropped like anything else, and so is the frame behind it.
fn stopping(shared: &Shared) -> bool {
    matches!(shared.phase, Phase::Stopping)
}

fn fail_shared(shared: &Arc<Mutex<Shared>>, error: impl Into<String>) -> bool {
    shared
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .fail(error);
    false
}

fn apply_ready(title: Option<String>, refresh_ms: Option<u64>, shared: &mut Shared) -> bool {
    if stopping(shared) {
        return true;
    }
    if shared.ready {
        shared.fail("plugin sent `ready` more than once");
        return false;
    }
    if !matches!(shared.phase, Phase::Starting) {
        return false;
    }
    shared.ready = true;
    shared.phase = Phase::Running;
    shared.title = title;
    shared.refresh = refresh_ms.map_or(DEFAULT_REFRESH, |milliseconds| {
        Duration::from_millis(milliseconds.clamp(16, 60_000))
    });
    shared.changed();
    true
}

fn apply_frame(frame: WireFrame, shared: &mut Shared) -> bool {
    shared.frame_sequence = shared.frame_sequence.wrapping_add(1);
    if shared
        .frame
        .as_ref()
        .is_some_and(|previous| frame.revision <= previous.revision)
    {
        return true;
    }
    shared.frame = Some(Arc::new(frame));
    shared.notice = None;
    shared.changed();
    true
}

fn validate_frame(
    title: Option<&str>,
    counter: Option<&str>,
    lines: &[WireLine],
    bindings: &[WireBinding],
    input: &InputPolicy,
) -> Result<(), String> {
    if lines.len() > MAX_FRAME_LINES {
        return Err(format!("plugin frame exceeds {MAX_FRAME_LINES} lines"));
    }
    if bindings.len() > MAX_BINDINGS {
        return Err(format!("plugin frame exceeds {MAX_BINDINGS} bindings"));
    }
    if input.keys.len() > MAX_INPUT_KEYS {
        return Err(format!("plugin frame exceeds {MAX_INPUT_KEYS} input keys"));
    }

    let mut retained = 0usize;
    if let Some(title) = title {
        validate_text(title, "frame title", MAX_TITLE_BYTES)?;
        add_retained(&mut retained, title.len())?;
    }
    if let Some(counter) = counter {
        validate_text(counter, "frame counter", MAX_COUNTER_BYTES)?;
        add_retained(&mut retained, counter.len())?;
    }

    let mut spans = 0usize;
    let theme = crate::theme::Theme::default();
    for line in lines {
        spans = spans
            .checked_add(line.spans.len())
            .ok_or_else(|| "plugin frame span count overflowed".to_string())?;
        if spans > MAX_FRAME_SPANS {
            return Err(format!("plugin frame exceeds {MAX_FRAME_SPANS} spans"));
        }
        for span in &line.spans {
            validate_display_text(&span.text, "span text")?;
            add_retained(&mut retained, span.text.len())?;
            for (field, color) in [("foreground", &span.fg), ("background", &span.bg)] {
                if let Some(color) = color {
                    validate_text(color, field, MAX_COLOR_BYTES)?;
                    if wire_color(color, &theme).is_none() {
                        return Err(format!(
                            "plugin span has an unknown {field} colour `{color}`"
                        ));
                    }
                    add_retained(&mut retained, color.len())?;
                }
            }
        }
    }

    for binding in bindings {
        validate_text(&binding.key, "binding key", MAX_BINDING_KEY_BYTES)?;
        validate_text(&binding.action, "binding action", MAX_BINDING_ACTION_BYTES)?;
        if binding_mentions_host_key(&binding.key, input.capture) {
            return Err(format!(
                "plugin binding `{}` is owned by the Mirador shell",
                binding.key
            ));
        }
        add_retained(&mut retained, binding.key.len())?;
        add_retained(&mut retained, binding.action.len())?;
    }

    for key in &input.keys {
        validate_text(key, "input key", MAX_BINDING_KEY_BYTES)?;
        if is_ctrl_c_chord(key) || (!input.capture && host_owned_chord(key)) {
            return Err(format!(
                "plugin input key `{key}` is owned by the Mirador shell"
            ));
        }
        add_retained(&mut retained, key.len())?;
    }
    Ok(())
}

fn validate_text(text: &str, field: &str, max_bytes: usize) -> Result<(), String> {
    if text.len() > max_bytes {
        return Err(format!("plugin {field} exceeds {max_bytes} UTF-8 bytes"));
    }
    validate_display_text(text, field)
}

fn validate_display_text(text: &str, field: &str) -> Result<(), String> {
    if text.chars().any(char::is_control) {
        return Err(format!(
            "plugin {field} contains a terminal control character"
        ));
    }
    Ok(())
}

fn add_retained(total: &mut usize, bytes: usize) -> Result<(), String> {
    *total = total
        .checked_add(bytes)
        .ok_or_else(|| "plugin frame text size overflowed".to_string())?;
    if *total > MAX_RETAINED_FRAME_TEXT_BYTES {
        return Err(format!(
            "plugin frame retains more than {MAX_RETAINED_FRAME_TEXT_BYTES} UTF-8 text bytes"
        ));
    }
    Ok(())
}

fn stderr_loop(
    mut stderr: impl io::Read,
    shared: &Arc<Mutex<Shared>>,
    supervisor: &mpsc::Sender<SupervisorCommand>,
) {
    let mut buffer = [0u8; 1024];
    let mut decoder = StderrDecoder::default();
    let mut window_started = Instant::now();
    let mut window_bytes = 0usize;
    loop {
        let read = match stderr.read(&mut buffer) {
            Ok(0) => {
                append_stderr(shared, &decoder.decode(&[], true));
                return;
            }
            Ok(read) => read,
            Err(error) => {
                shared
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .fail(format!("reading plugin stderr: {error}"));
                let _ = supervisor.send(SupervisorCommand::Abort);
                return;
            }
        };
        let now = Instant::now();
        if now.saturating_duration_since(window_started) >= MESSAGE_RATE_WINDOW {
            window_started = now;
            window_bytes = 0;
        }
        window_bytes = window_bytes.saturating_add(read);
        if window_bytes > MAX_STDERR_BYTES_PER_WINDOW {
            shared
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .fail(format!(
                    "plugin exceeded {MAX_STDERR_BYTES_PER_WINDOW} stderr bytes in one second"
                ));
            let _ = supervisor.send(SupervisorCommand::Abort);
            return;
        }

        append_stderr(shared, &decoder.decode(&buffer[..read], false));
    }
}

#[derive(Default)]
struct StderrDecoder {
    pending: Vec<u8>,
}

impl StderrDecoder {
    /// Decode complete UTF-8 while retaining at most one incomplete code point
    /// for the next fixed-size read. Invalid bytes still become replacement
    /// characters; valid text split exactly at 1 KiB no longer does.
    fn decode(&mut self, bytes: &[u8], eof: bool) -> String {
        self.pending.extend_from_slice(bytes);
        let mut output = String::new();
        loop {
            match std::str::from_utf8(&self.pending) {
                Ok(valid) => {
                    sanitise_stderr(valid, &mut output);
                    self.pending.clear();
                    break;
                }
                Err(error) => {
                    let valid = error.valid_up_to();
                    if valid > 0 {
                        // `valid_up_to` is guaranteed to be a UTF-8 boundary.
                        let prefix =
                            std::str::from_utf8(&self.pending[..valid]).unwrap_or_default();
                        sanitise_stderr(prefix, &mut output);
                        self.pending.drain(..valid);
                    }
                    match error.error_len() {
                        Some(length) => {
                            output.push('\u{fffd}');
                            self.pending.drain(..length);
                        }
                        None if eof => {
                            output.push('\u{fffd}');
                            self.pending.clear();
                            break;
                        }
                        None => break,
                    }
                }
            }
        }
        output
    }
}

fn sanitise_stderr(text: &str, output: &mut String) {
    output.extend(text.chars().map(|character| match character {
        '\n' | '\r' => '\n',
        '\t' => ' ',
        control if control.is_control() => '\u{fffd}',
        printable => printable,
    }));
}

fn append_stderr(shared: &Arc<Mutex<Shared>>, text: &str) {
    if text.is_empty() {
        return;
    }
    let mut shared = shared
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    shared.stderr.push_str(text);
    if shared.stderr.len() > MAX_STDERR_BYTES {
        let excess = shared.stderr.len() - MAX_STDERR_BYTES;
        let mut boundary = excess;
        while boundary < shared.stderr.len() && !shared.stderr.is_char_boundary(boundary) {
            boundary += 1;
        }
        shared.stderr.drain(..boundary);
    }
    // Stderr is visible before the first frame and after a terminal failure.
    // While a healthy frame is on screen, retaining diagnostics is enough; a
    // generation bump per chunk would force redraws for text not being drawn.
    if shared.frame.is_none()
        || matches!(
            shared.phase,
            Phase::Exited(_) | Phase::Failed(_) | Phase::Stopping
        )
    {
        shared.changed();
    }
}

fn supervisor_loop(
    child: &mut Child,
    commands: &Receiver<SupervisorCommand>,
    shared: &Arc<Mutex<Shared>>,
    startup_deadline: Instant,
) {
    let mut deadline = None;
    let mut output_closed = None;
    let mut disconnected = false;
    loop {
        match child.try_wait() {
            Ok(Some(status)) => {
                let mut shared = shared
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner);
                if !matches!(shared.phase, Phase::Stopping | Phase::Failed(_)) {
                    if shared.ready {
                        shared.phase = Phase::Exited(status.to_string());
                    } else {
                        shared.phase =
                            Phase::Failed(format!("plugin exited before `ready`: {status}"));
                    }
                    shared.changed();
                }
                return;
            }
            Ok(None) => {}
            Err(error) => {
                shared
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .fail(format!("checking plugin process: {error}"));
                terminate_child(child);
                return;
            }
        }

        let startup_timed_out = expire_startup(shared, Instant::now(), startup_deadline);
        if startup_timed_out {
            terminate_child(child);
            return;
        }

        if output_closed.is_some_and(|when| Instant::now() >= when) {
            let should_fail = {
                let mut shared = shared
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner);
                if matches!(shared.phase, Phase::Stopping | Phase::Failed(_)) {
                    false
                } else {
                    shared.fail("plugin closed stdout without exiting");
                    true
                }
            };
            if should_fail {
                terminate_child(child);
                return;
            }
            output_closed = None;
        }

        if deadline.is_some_and(|when| Instant::now() >= when) {
            terminate_child(child);
            return;
        }

        let now = Instant::now();
        let startup = (!shared
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .ready)
            .then_some(startup_deadline);
        let until = [startup, deadline, output_closed]
            .into_iter()
            .flatten()
            .map(|when| when.saturating_duration_since(now))
            .min()
            .unwrap_or(PROCESS_STATUS_POLL)
            .min(PROCESS_STATUS_POLL);
        if disconnected {
            thread::sleep(until);
            continue;
        }
        match commands.recv_timeout(until) {
            Ok(SupervisorCommand::Abort) => {
                terminate_child(child);
                return;
            }
            Ok(SupervisorCommand::Shutdown) => {
                deadline = Some(Instant::now() + SHUTDOWN_GRACE);
                let mut shared = shared
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner);
                if !matches!(shared.phase, Phase::Exited(_) | Phase::Failed(_)) {
                    shared.phase = Phase::Stopping;
                    shared.changed();
                }
            }
            Ok(SupervisorCommand::OutputClosed) => {
                output_closed = Some(Instant::now() + OUTPUT_CLOSE_GRACE);
            }
            Err(RecvTimeoutError::Disconnected) if !disconnected => {
                disconnected = true;
                deadline = Some(Instant::now() + SHUTDOWN_GRACE);
            }
            Err(RecvTimeoutError::Timeout | RecvTimeoutError::Disconnected) => {}
        }
    }
}

fn terminate_child(child: &mut Child) {
    let _ = child.kill();
    let _ = child.wait();
}

fn expire_startup(shared: &Arc<Mutex<Shared>>, now: Instant, deadline: Instant) -> bool {
    let mut shared = shared
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    if now >= deadline && !shared.ready && matches!(shared.phase, Phase::Starting) {
        shared.fail(format!(
            "plugin did not send `ready` within {} seconds",
            STARTUP_TIMEOUT.as_secs()
        ));
        true
    } else {
        false
    }
}

#[cfg(test)]
mod tests {
    use std::process::Stdio;

    use ratatui::crossterm::event::{KeyCode, KeyEvent};

    use super::*;
    use crate::panel::{KeyOutcome, Panel};
    use crate::plugin::tests::detached_panel;
    use crate::plugin::{InputPolicy, PluginPanel, WireBinding, WireSpan};

    const WAITING_CHILD_TEST: &str =
        "plugin::process::tests::child_waits_for_the_parent_to_terminate_it";

    fn waiting_child() -> Child {
        let executable = std::env::current_exe().expect("test executable has a path");
        Command::new(executable)
            .args(["--ignored", "--exact", WAITING_CHILD_TEST, "--quiet"])
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn()
            .expect("waiting test child starts")
    }

    fn supervised_waiting_runtime(
        shared: &Arc<Mutex<Shared>>,
    ) -> (Runtime, Receiver<HostMessage>, Arc<AtomicBool>) {
        let mut child = waiting_child();
        let (events, received) = mpsc::sync_channel(EVENT_QUEUE);
        let (supervisor, commands) = mpsc::channel();
        let finished = Arc::new(AtomicBool::new(false));
        let thread_finished = Arc::clone(&finished);
        let supervisor_shared = Arc::clone(shared);
        let supervisor_thread = thread::spawn(move || {
            supervisor_loop(
                &mut child,
                &commands,
                &supervisor_shared,
                Instant::now() + STARTUP_TIMEOUT,
            );
            thread_finished.store(true, Ordering::Release);
        });
        (
            Runtime {
                events: Some(events),
                stopping: Arc::new(AtomicBool::new(false)),
                tick_pending: Arc::new(AtomicBool::new(false)),
                supervisor,
                supervisor_thread: Some(supervisor_thread),
            },
            received,
            finished,
        )
    }

    #[test]
    #[ignore = "helper process for portable plugin lifecycle tests"]
    fn child_waits_for_the_parent_to_terminate_it() {
        thread::sleep(Duration::from_secs(30));
    }

    #[test]
    fn normal_host_shutdown_waits_for_the_supervisor_to_reap_its_child() {
        let shared = Arc::new(Mutex::new(Shared::starting()));
        let (runtime, _received, finished) = supervised_waiting_runtime(&shared);
        let mut panel = PluginPanel::new(PluginConfig {
            id: "shutdown-test".into(),
            command: vec!["mirador-plugin-shutdown-test-command-does-not-exist".into()],
            config: toml::Table::new(),
        });
        panel.shared = shared;
        panel.runtime = Some(runtime);
        let started = Instant::now();

        panel.shutdown();
        assert!(
            panel.runtime.is_none(),
            "the panel retained a supervisor after shutdown"
        );
        assert!(
            started.elapsed() >= SHUTDOWN_GRACE,
            "the configured child was not given its cleanup grace"
        );
        assert!(
            started.elapsed() < SUPERVISOR_JOIN_GRACE + SUPERVISOR_ABORT_GRACE,
            "normal shutdown exceeded the host's bounded wait"
        );
        assert!(
            finished.load(Ordering::Acquire),
            "shutdown returned before kill/wait completed"
        );
    }

    #[test]
    fn runtime_drop_enforces_the_same_bounded_child_cleanup() {
        let shared = Arc::new(Mutex::new(Shared::starting()));
        let (runtime, _received, finished) = supervised_waiting_runtime(&shared);
        let started = Instant::now();

        drop(runtime);

        assert!(started.elapsed() >= SHUTDOWN_GRACE);
        assert!(started.elapsed() < Duration::from_secs(2));
        assert!(
            finished.load(Ordering::Acquire),
            "dropping the runtime orphaned its child supervisor"
        );
    }

    #[test]
    fn the_startup_timeout_terminates_and_reaps_an_unready_child() {
        let mut child = waiting_child();
        let (_commands, received) = mpsc::channel();
        let shared = Arc::new(Mutex::new(Shared::starting()));
        let started = Instant::now();

        supervisor_loop(
            &mut child,
            &received,
            &shared,
            Instant::now() + Duration::from_millis(25),
        );

        assert!(matches!(
            shared.lock().unwrap().phase,
            Phase::Failed(ref error) if error.contains("did not send `ready`")
        ));
        assert!(
            started.elapsed() < Duration::from_secs(2),
            "the unready helper was waited out instead of terminated"
        );
    }

    #[test]
    fn closed_output_gets_a_short_grace_then_terminates_the_child() {
        let mut child = waiting_child();
        let (commands, received) = mpsc::channel();
        let mut state = Shared::starting();
        state.ready = true;
        state.phase = Phase::Running;
        let shared = Arc::new(Mutex::new(state));
        commands.send(SupervisorCommand::OutputClosed).unwrap();
        let started = Instant::now();

        supervisor_loop(
            &mut child,
            &received,
            &shared,
            Instant::now() + STARTUP_TIMEOUT,
        );

        assert!(started.elapsed() >= OUTPUT_CLOSE_GRACE);
        assert!(matches!(
            shared.lock().unwrap().phase,
            Phase::Failed(ref error) if error.contains("closed stdout")
        ));
        assert!(
            started.elapsed() < Duration::from_secs(2),
            "the output-closed helper was waited out instead of terminated"
        );
    }

    #[test]
    fn a_child_that_exits_is_recorded_without_waiting_for_a_command() {
        let executable = std::env::current_exe().expect("test executable has a path");
        let mut child = Command::new(executable)
            .arg("--help")
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn()
            .expect("short-lived test child starts");
        let (_commands, received) = mpsc::channel();
        let mut state = Shared::starting();
        state.ready = true;
        state.phase = Phase::Running;
        let shared = Arc::new(Mutex::new(state));

        supervisor_loop(
            &mut child,
            &received,
            &shared,
            Instant::now() + STARTUP_TIMEOUT,
        );

        assert!(matches!(shared.lock().unwrap().phase, Phase::Exited(_)));
    }

    #[test]
    fn inbound_message_rate_has_a_rolling_bound() {
        let start = Instant::now();
        let mut rate = MessageRate::default();
        for _ in 0..MAX_MESSAGES_PER_WINDOW {
            assert!(rate.accepts(start, 1));
        }
        assert!(!rate.accepts(start, 1));
        assert!(rate.accepts(start + MESSAGE_RATE_WINDOW, 1));

        let mut bytes = MessageRate::default();
        assert!(bytes.accepts(start, MAX_STDOUT_BYTES_PER_WINDOW));
        assert!(!bytes.accepts(start, 1));
    }

    #[test]
    fn startup_deadline_fails_only_an_unready_process() {
        let now = Instant::now();
        let waiting = Arc::new(Mutex::new(Shared::starting()));
        assert!(expire_startup(&waiting, now, now));
        assert!(matches!(
            waiting.lock().unwrap().phase,
            Phase::Failed(ref error) if error.contains("within 5 seconds")
        ));

        let ready = Arc::new(Mutex::new(Shared::starting()));
        ready.lock().unwrap().ready = true;
        assert!(!expire_startup(&ready, now, now));
    }

    /// `shutdown` moves the supervisor to `Stopping` at once, but the plugin
    /// hears it only once the writer has flushed it, so a plugin drawing every
    /// 33 ms nearly always has a frame in flight. Every kind of message read in
    /// that window used to be refused, and a refusal is an abort: the child was
    /// killed in milliseconds, without the 300 ms the protocol promises for
    /// cleanup. Dropped, each must leave the phase where it was and draw
    /// nothing.
    #[test]
    fn a_message_during_the_shutdown_grace_is_dropped_rather_than_fatal() {
        let ready = || PluginMessage::Ready {
            protocol: PROTOCOL_VERSION,
            title: None,
            refresh_ms: None,
        };
        let frame = || PluginMessage::Frame {
            revision: 1,
            title: None,
            counter: None,
            lines: vec![WireLine {
                spans: vec![WireSpan::plain("saving")],
            }],
            bindings: Vec::new(),
            input: InputPolicy::default(),
            cursor: None,
        };
        let stopping = || {
            let shared = Arc::new(Mutex::new(Shared::starting()));
            assert!(apply_message(ready(), &shared));
            shared.lock().unwrap().phase = Phase::Stopping;
            shared
        };

        let shared = stopping();
        let generation = shared.lock().unwrap().generation;
        for (what, message) in [
            ("a frame", frame()),
            (
                "an error",
                PluginMessage::Error {
                    message: "saving".into(),
                    fatal: false,
                },
            ),
            (
                "a fatal error",
                PluginMessage::Error {
                    message: "gone".into(),
                    fatal: true,
                },
            ),
            (
                "a watch event",
                PluginMessage::Watch {
                    text: "saved".into(),
                },
            ),
            ("a second ready", ready()),
        ] {
            assert!(
                apply_message(message, &shared),
                "{what} cut the grace short"
            );
            let shared = shared.lock().unwrap();
            assert_eq!(shared.phase, Phase::Stopping, "after {what}");
            assert_eq!(shared.generation, generation, "{what} was drawn");
        }
        let shared = shared.lock().unwrap();
        assert!(shared.frame.is_none() && shared.notice.is_none() && shared.watch.is_empty());
        drop(shared);

        // Shut down before it ever said `ready`: the `ready` and the frame
        // behind it were both in flight.
        let shared = Arc::new(Mutex::new(Shared::starting()));
        shared.lock().unwrap().phase = Phase::Stopping;
        assert!(apply_message(ready(), &shared), "a late ready");
        assert!(apply_message(frame(), &shared), "the frame behind it");
        assert_eq!(shared.lock().unwrap().phase, Phase::Stopping);

        // The grace is for a plugin that is stopping, not for one that is
        // broken: a malformed message is still refused.
        assert!(!apply_message(
            PluginMessage::Watch {
                text: "two\nlines".into()
            },
            &stopping()
        ));
    }

    #[test]
    fn concurrent_workers_preserve_the_first_terminal_error() {
        let mut shared = Shared::starting();
        shared.fail("invalid plugin message");
        let generation = shared.generation;
        shared.fail("writing to plugin: broken pipe");
        assert_eq!(shared.generation, generation);
        assert_eq!(shared.phase, Phase::Failed("invalid plugin message".into()));
    }

    #[test]
    fn stderr_is_sanitised_retained_and_rate_bounded() {
        let shared = Arc::new(Mutex::new(Shared::starting()));
        let (supervisor, _commands) = mpsc::channel();
        stderr_loop("safe\u{1b}[2J".as_bytes(), &shared, &supervisor);
        assert_eq!(shared.lock().unwrap().stderr, "safe�[2J");

        let shared = Arc::new(Mutex::new(Shared::starting()));
        stderr_loop(
            &vec![b'x'; MAX_STDERR_BYTES_PER_WINDOW + 1][..],
            &shared,
            &supervisor,
        );
        assert!(matches!(
            shared.lock().unwrap().phase,
            Phase::Failed(ref error) if error.contains("stderr bytes")
        ));
    }

    #[test]
    fn stderr_decoder_preserves_utf8_split_at_a_read_boundary() {
        let encoded = "é".as_bytes();
        let mut decoder = StderrDecoder::default();
        assert_eq!(decoder.decode(&encoded[..1], false), "");
        assert_eq!(decoder.decode(&encoded[1..], false), "é");

        let mut incomplete = StderrDecoder::default();
        assert_eq!(incomplete.decode(&encoded[..1], true), "\u{fffd}");
    }

    /// A child's stdin that takes the first write and then holds it, the way
    /// a plugin that has stopped reading holds the host's writer, until the
    /// test lets go — which is the plugin reading again inside its grace.
    struct HeldPipe {
        entered: Option<mpsc::Sender<()>>,
        release: Option<Receiver<()>>,
        written: Arc<Mutex<Vec<u8>>>,
    }

    impl Write for HeldPipe {
        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
            if let Some(entered) = self.entered.take() {
                let _ = entered.send(());
            }
            if let Some(release) = self.release.take() {
                let _ = release.recv();
            }
            self.written
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .extend_from_slice(bytes);
            Ok(bytes.len())
        }

        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    fn hello() -> HostMessage {
        HostMessage::Hello {
            protocol: PROTOCOL_VERSION,
            host_version: "test",
            plugin: "held".into(),
            config: serde_json::Value::Null,
            cwd: String::new(),
        }
    }

    fn key_x() -> HostMessage {
        HostMessage::Key {
            key: "x".into(),
            code: "char".into(),
            text: Some("x".into()),
            modifiers: Vec::new(),
        }
    }

    fn lines_of(written: &Arc<Mutex<Vec<u8>>>) -> Vec<String> {
        let bytes = written
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone();
        String::from_utf8(bytes)
            .expect("the host writes UTF-8")
            .lines()
            .map(str::to_string)
            .collect()
    }

    /// `shutdown` went into the same bounded queue as everything else, by
    /// `try_send`, so a plugin that had fallen 256 messages behind was never
    /// told — the queue was full, the message was dropped, and the child was
    /// killed at the end of its grace without the one word the protocol
    /// promises it. Here the child stops reading mid-`hello`, the queue fills
    /// behind it — one tick, since ticks wait one at a time, and keys in
    /// every other place — the host shuts down, and then the child reads
    /// again: what follows the `hello` is `shutdown`, not the queue.
    /// `HeldPipe` has no buffer, so this is the host's queue alone; a real
    /// pipe hands over what it already holds first, as the protocol
    /// document says.
    #[test]
    fn shutdown_reaches_the_plugin_past_a_full_queue() {
        let shared = Arc::new(Mutex::new(Shared::starting()));
        let (entered_tx, entered) = mpsc::channel();
        let (release, release_rx) = mpsc::channel();
        let written = Arc::new(Mutex::new(Vec::new()));
        let pipe = HeldPipe {
            entered: Some(entered_tx),
            release: Some(release_rx),
            written: Arc::clone(&written),
        };
        let (mut runtime, writer) = runtime_writing_to(pipe, &shared);

        runtime.send(hello()).expect("the hello is queued");
        entered.recv().expect("the writer takes the hello");
        runtime.send_tick();
        let mut queued = 1;
        while runtime.send(key_x()).is_ok() {
            queued += 1;
        }
        assert_eq!(queued, EVENT_QUEUE, "the queue fills behind the held write");

        runtime.shutdown();
        drop(release);
        writer.join().expect("the writer finishes");

        let lines = lines_of(&written);
        assert_eq!(
            lines.last().map(String::as_str),
            Some(r#"{"type":"shutdown"}"#),
            "the plugin was never told to stop; it read {} lines, the last {:?}",
            lines.len(),
            lines.last()
        );
        assert!(lines[0].contains(r#""type":"hello""#), "{lines:?}");
        assert_eq!(
            lines.len(),
            2,
            "what was still queued is skipped, so the shutdown is not stuck behind it"
        );
    }

    /// The two ends of the same rule. A `hello` still queued when the host
    /// stops is written before the `shutdown`, never replaced by it — the
    /// protocol lets `shutdown` follow `hello` during startup, not stand in
    /// for it — and the tick and input behind it are skipped. And a writer
    /// waiting on an empty queue when the host stops still says `shutdown`
    /// on its way out, since nothing is left in the queue to carry it.
    #[test]
    fn shutdown_follows_a_queued_hello_and_ends_an_empty_queue() {
        let shared = Arc::new(Mutex::new(Shared::starting()));
        let (supervisor, _commands) = mpsc::channel();
        let stopping = AtomicBool::new(true);
        let tick_pending = AtomicBool::new(true);

        let (events, received) = mpsc::sync_channel(EVENT_QUEUE);
        events.try_send(hello()).expect("room");
        events.try_send(HostMessage::Tick).expect("room");
        events.try_send(key_x()).expect("room");
        drop(events);
        let mut written = Vec::new();
        writer_loop(
            &mut written,
            &received,
            &stopping,
            &tick_pending,
            &shared,
            &supervisor,
        );
        let lines: Vec<_> = String::from_utf8(written)
            .expect("UTF-8")
            .lines()
            .map(str::to_string)
            .collect();
        assert_eq!(lines.len(), 2, "{lines:?}");
        assert!(lines[0].contains(r#""type":"hello""#), "{lines:?}");
        assert_eq!(lines[1], r#"{"type":"shutdown"}"#);

        let (events, received) = mpsc::sync_channel::<HostMessage>(EVENT_QUEUE);
        drop(events);
        let mut written = Vec::new();
        writer_loop(
            &mut written,
            &received,
            &stopping,
            &tick_pending,
            &shared,
            &supervisor,
        );
        assert_eq!(written, b"{\"type\":\"shutdown\"}\n");
    }

    /// A child's stdin read as fast as it is written: each message is handed
    /// to the test as it lands, so the test waits on the writer rather than
    /// sleeping for it.
    struct ReadingPipe {
        partial: Vec<u8>,
        lines: mpsc::Sender<String>,
    }

    impl Write for ReadingPipe {
        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
            for &byte in bytes {
                if byte == b'\n' {
                    let line = String::from_utf8(std::mem::take(&mut self.partial))
                        .expect("the host writes UTF-8");
                    let _ = self.lines.send(line);
                } else {
                    self.partial.push(byte);
                }
            }
            Ok(bytes.len())
        }

        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    /// Everything on the host's side of a plugin's stdin: a runtime whose
    /// queue a real `writer_loop` drains into `pipe`, sharing its flags the
    /// way `spawn_process` wires them. There is no child, so there is no
    /// supervisor to wait for when it shuts down.
    fn runtime_writing_to(
        pipe: impl Write + Send + 'static,
        shared: &Arc<Mutex<Shared>>,
    ) -> (Runtime, JoinHandle<()>) {
        let (events, received) = mpsc::sync_channel(EVENT_QUEUE);
        let (supervisor, _commands) = mpsc::channel();
        let stopping = Arc::new(AtomicBool::new(false));
        let tick_pending = Arc::new(AtomicBool::new(false));
        let writer = {
            let stopping = Arc::clone(&stopping);
            let tick_pending = Arc::clone(&tick_pending);
            let shared = Arc::clone(shared);
            let supervisor = supervisor.clone();
            thread::spawn(move || {
                writer_loop(
                    pipe,
                    &received,
                    &stopping,
                    &tick_pending,
                    &shared,
                    &supervisor,
                );
            })
        };
        let runtime = Runtime {
            events: Some(events),
            stopping,
            tick_pending,
            supervisor,
            supervisor_thread: None,
        };
        (runtime, writer)
    }

    fn capturing() -> InputPolicy {
        InputPolicy {
            capture: true,
            ..InputPolicy::default()
        }
    }

    /// A running external panel that captures keys, over
    /// [`runtime_writing_to`], so its `tick` goes all the way to the pipe.
    fn panel_writing_to(pipe: impl Write + Send + 'static) -> (PluginPanel, JoinHandle<()>) {
        let mut panel = detached_panel(capturing());
        let (runtime, writer) = runtime_writing_to(pipe, &panel.shared);
        panel.runtime = Some(runtime);
        (panel, writer)
    }

    fn press_x(panel: &mut PluginPanel) -> KeyOutcome {
        panel.handle_key(KeyEvent::from(KeyCode::Char('x')))
    }

    fn notice_of(panel: &PluginPanel) -> Option<String> {
        panel
            .shared
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .notice
            .clone()
    }

    /// A plugin that stopped reading had its queue filled by the host's own
    /// ticks, one every poll — all 256 places in eight seconds at 33 ms on a
    /// focused panel capturing input, once the pipe behind it was full, and
    /// in about a minute at the default 250 ms otherwise — after which the
    /// panel said `plugin input queue is full` about input nobody had typed,
    /// and dropped every key that was. A tick is a cue, not a count, so
    /// however many refreshes pass, one waits, and the rest of the queue is
    /// left for the reader's input. The stub's queue has no writer behind
    /// it, which is a plugin that has stopped reading with its pipe full.
    #[test]
    fn a_plugin_that_stops_reading_has_one_tick_waiting_and_room_for_keys() {
        let mut panel = detached_panel(capturing());
        let (runtime, received) = Runtime::stub();
        panel.runtime = Some(runtime);

        for _ in 0..1_000 {
            panel.tick();
        }
        let pressed = press_x(&mut panel);

        let queued: Vec<HostMessage> = received.try_iter().collect();
        let ticks = queued
            .iter()
            .filter(|message| matches!(message, HostMessage::Tick))
            .count();
        assert_eq!(ticks, 1, "a thousand refreshes left {ticks} ticks waiting");
        assert!(
            matches!(queued.last(), Some(HostMessage::Key { key, .. }) if key == "x"),
            "the key did not fit behind the ticks: {} queued, the last {:?}",
            queued.len(),
            queued.last()
        );
        assert_eq!(
            notice_of(&panel),
            None,
            "the panel blamed input for the host's own ticks"
        );
        assert_eq!(pressed, KeyOutcome::Consumed);
        panel.shutdown();
    }

    /// A tick that finds the queue full of the reader's own input is dropped
    /// without a word: it is not input, and the next refresh offers another.
    /// And coalescing must not make that drop permanent — once the plugin
    /// reads one message, the next refresh's tick takes the room.
    #[test]
    fn a_tick_that_finds_the_queue_full_blames_nobody_and_comes_back() {
        let mut panel = detached_panel(capturing());
        let (runtime, received) = Runtime::stub();
        panel.runtime = Some(runtime);
        for _ in 0..EVENT_QUEUE {
            assert_eq!(press_x(&mut panel), KeyOutcome::Consumed);
        }
        assert_eq!(notice_of(&panel), None, "every key fitted");

        panel.tick();
        assert_eq!(
            notice_of(&panel),
            None,
            "a tick with no room is not input dropped"
        );

        received.try_recv().expect("the plugin reads one key");
        panel.tick();
        let queued: Vec<HostMessage> = received.try_iter().collect();
        assert!(
            matches!(queued.last(), Some(HostMessage::Tick)),
            "the tick refused for want of room never came back: {} queued, the last {:?}",
            queued.len(),
            queued.last()
        );
        panel.shutdown();
    }

    /// The writer clears the pending tick *before* writing it, not after. A
    /// plugin slow to read holds the writer inside that write while further
    /// refreshes raise further ticks; cleared afterwards, the flag would
    /// still say one was waiting, every one of them would be folded into a
    /// tick already on its way out, and a plugin that caught up would find
    /// nothing behind it. Cleared first, the next is queued and the rest
    /// join it: one more tick, not none and not a backlog.
    #[test]
    fn a_tick_raised_during_a_write_is_delivered_once_afterwards() {
        let (entered_tx, entered) = mpsc::channel();
        let (release, release_rx) = mpsc::channel();
        let written = Arc::new(Mutex::new(Vec::new()));
        let (mut panel, writer) = panel_writing_to(HeldPipe {
            entered: Some(entered_tx),
            release: Some(release_rx),
            written: Arc::clone(&written),
        });

        panel.tick();
        entered.recv().expect("the writer takes the first tick");
        for _ in 0..10 {
            panel.tick();
        }
        drop(release);
        // Close the queue without stopping, so the writer writes what is
        // left in it and leaves, rather than swapping it for `shutdown`.
        let runtime = panel.runtime.as_mut().expect("the panel has a runtime");
        drop(runtime.events.take());
        writer.join().expect("the writer finishes");

        assert_eq!(
            lines_of(&written),
            [r#"{"type":"tick"}"#; 2],
            "ten refreshes during one held write"
        );
    }

    /// The other side of coalescing: a plugin that reads as fast as the host
    /// writes is not starved by it. Each tick is written before the next
    /// refresh, so each refresh finds none waiting and sends its own.
    ///
    /// This catches a writer that never clears the flag. One that clears it
    /// only after the write is left to
    /// `a_tick_raised_during_a_write_is_delivered_once_afterwards`: here the
    /// line reaches the test inside that write and the clear follows a few
    /// instructions later, a race this test almost never wins. Broken that
    /// way, it passed thirty runs out of thirty.
    #[test]
    fn a_plugin_that_keeps_up_gets_a_tick_every_refresh() {
        let (lines_tx, lines) = mpsc::channel();
        let (mut panel, writer) = panel_writing_to(ReadingPipe {
            partial: Vec::new(),
            lines: lines_tx,
        });

        for refresh in 0..20 {
            panel.tick();
            assert_eq!(
                lines.recv_timeout(Duration::from_secs(5)).as_deref(),
                Ok(r#"{"type":"tick"}"#),
                "refresh {refresh} sent no tick"
            );
        }
        panel.shutdown();
        writer.join().expect("the writer finishes");
        assert_eq!(
            lines.try_iter().collect::<Vec<_>>(),
            [r#"{"type":"shutdown"}"#],
            "anything but the shutdown after the last tick"
        );
    }

    #[test]
    fn outbound_messages_obey_the_same_byte_limit() {
        let event = HostMessage::Paste {
            text: "x".repeat(MAX_MESSAGE_BYTES),
        };
        let error = write_host_message(Vec::new(), &event).unwrap_err();
        assert_eq!(error.kind(), io::ErrorKind::InvalidData);
    }

    #[test]
    fn frames_cannot_smuggle_terminal_controls_or_retain_unbounded_text() {
        let controls = [WireLine {
            spans: vec![WireSpan::plain("safe\u{1b}[2Jnot safe")],
        }];
        let error =
            validate_frame(None, None, &controls, &[], &InputPolicy::default()).unwrap_err();
        assert!(error.contains("control character"), "{error}");

        let oversized = [WireLine {
            spans: vec![WireSpan::plain(
                "x".repeat(MAX_RETAINED_FRAME_TEXT_BYTES + 1),
            )],
        }];
        let error =
            validate_frame(None, None, &oversized, &[], &InputPolicy::default()).unwrap_err();
        assert!(error.contains("retains more"), "{error}");
    }

    #[test]
    fn passive_frames_cannot_advertise_shell_bindings() {
        let input = InputPolicy {
            keys: vec!["q".into()],
            ..InputPolicy::default()
        };
        let error = validate_frame(None, None, &[], &[], &input).unwrap_err();
        assert!(error.contains("owned by the Mirador shell"), "{error}");

        let bindings = [WireBinding {
            key: "Ctrl+C / Ctrl+X".into(),
            action: "interrupt".into(),
            primary: true,
        }];
        let capturing = InputPolicy {
            capture: true,
            ..InputPolicy::default()
        };
        let error = validate_frame(None, None, &[], &bindings, &capturing).unwrap_err();
        assert!(error.contains("owned by the Mirador shell"), "{error}");
    }
}