shep 0.1.2

//! The lookout's state and its reducer: `Msg` in, `Effect` out.
//!
//! Everything about this module is chosen so it can be tested without a
//! terminal, a runtime, a socket, or a sleep:
//!
//! - **No I/O.** [`App::update`] is a synchronous function of `&mut self` and
//!   one [`Msg`]. Work that has to happen outside comes back out as an
//!   [`Effect`] for the caller in `super::mod` to run.
//! - **No terminal types.** Nothing here imports `ratatui` beyond
//!   [`super::theme::Palette`] (which is a style value, not a widget) or
//!   `crossterm` at all — `super::input` maps a `KeyEvent` to a [`KeyPress`]
//!   before it reaches this module.
//! - **No clock.** Every `Instant` arrives on the message
//!   ([`Msg::Tick`], [`Msg::Snapshot`]). A test asserts on uptime arithmetic
//!   exactly rather than sleeping and hoping.
//!
//! **The flock map is keyed by sheep id, and the poll is the truth.** The bus
//! is lossy by construction (`tokio::sync::broadcast` drops for a lagging
//! subscriber), so a flock view built from events alone WILL drift.
//! [`Msg::Event`] upserts for latency; [`Msg::Snapshot`] replaces the whole map
//! and wins every conflict. The cursor this module carries is a **selected
//! sheep id**, not a row index — the flock map is replaced wholesale every two
//! seconds, so an index cursor would silently point at a different sheep the
//! moment an earlier row is deleted. [`App::reseat`] is what puts the
//! selection back on a real sheep after the map changes; the viewport offset
//! the flock table scrolls to is derived from the selection rather than
//! stored beside it ([`super::view::flock::scroll_offset`]).

use core::fmt;
use std::collections::BTreeMap;
use std::time::{Duration, Instant};

use shep_client::RequestError;
use shep_core::protocol::{
    BusEvent, Lamb, ProcessEventKind, ProcessInfo, Request, Response, SelectorSpec,
};
use shep_core::status::ProcStatus;

use super::theme::Palette;

/// Whether this lookout may act on a sheep.
///
/// Default is [`Self::ReadOnly`], per Rin's ruling, mirroring the
/// `allow_control` precedent spec §9 sets for whistle. Turned on by
/// `--allow-control` or by `lookout.allow_control` in the KV store.
///
/// **This is a fat-finger catch, not a security boundary.** lookout runs as the
/// operator's own process under the operator's own uid; anyone who can run it
/// can run `shep stop`. The gate exists so a keystroke in a dashboard someone
/// is reading does not become an action they did not intend.
// Every public item below this point is wired together by `super::mod`'s
// `lookout` and `run_ui` — the real caller for `App`, `Msg` and friends.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Control {
    /// Actions refuse. The default.
    ReadOnly,
    /// Actions are permitted. Three keys arm a confirm — `x` (stop), `R`
    /// (restart) and `L` (reload) — and Enter sends it.
    Allowed,
}

/// Which keymap is in force.
///
/// Held by the reducer and passed to [`super::input::map_key`] at the call
/// site, so the crossterm edge stays in one file and this module keeps
/// holding no terminal types.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum InputMode {
    /// The ordinary dashboard keys.
    Normal,
    /// The filter box is open and every printable key is text.
    Text,
}

/// The keys lookout binds, named by what they mean rather than by which key
/// produces them.
///
/// A plain enum, rather than `crossterm::event::KeyEvent`, so this module and
/// its tests never touch a terminal crate: `super::input::map_key` does the
/// translation at the edge.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum KeyPress {
    /// `q` in normal mode, or `Ctrl-C` in either.
    Quit,
    /// `Esc` in normal mode. Cancels an armed confirm if there is one, else
    /// clears the filter if there is one, else quits. The reducer decides,
    /// because the keymap cannot see any of those three states.
    Escape,
    /// `k` or `Up` — the selection moves up one row.
    SelectUp,
    /// `j` or `Down` — the selection moves down one row.
    SelectDown,
    /// `g` or `Home` — the first sheep in the flock.
    SelectFirst,
    /// `G` or `End` — the last one.
    SelectLast,
    /// `r` — poll now.
    Refresh,
    /// `x`, `R` or `L` — arms a confirm for the given verb, or refuses and
    /// says why. See [`ActionVerb`].
    Action(ActionVerb),
    /// `Enter` in normal mode. Sends an armed confirm; does nothing
    /// otherwise.
    Confirm,
    /// `/` — open the filter box, carrying whatever query is already set.
    FilterStart,
    /// One printable character typed into the box.
    FilterChar(char),
    /// `Backspace` in the box.
    FilterBackspace,
    /// `Enter` in the box: apply and leave.
    FilterApply,
    /// `Esc` in the box: clear the filter and leave.
    FilterAbandon,
}

/// Everything that can change the dashboard.
#[derive(Debug, Clone)]
pub enum Msg {
    /// A `Request::ListFlock` reply landed. `at` is when it was received, and
    /// becomes every row's uptime anchor.
    Snapshot {
        /// The flock as the shepherd reported it.
        rows: Vec<ProcessInfo>,
        /// When the reply was received.
        at: Instant,
    },
    /// One frame off the bus.
    Event(BusEvent),
    /// This client's own receiver fell behind and discarded frames — the
    /// local half of the drop problem, distinct from
    /// [`BusEvent::Dropped`], which is the shepherd's own queue.
    BusLagged {
        /// How many frames this process lost.
        count: u64,
    },
    /// The link task is re-dialling; `attempt` is 1-based.
    Retrying {
        /// Which attempt is in flight.
        attempt: u32,
    },
    /// The link task reconnected and re-subscribed.
    Relinked,
    /// The reconnect ladder is exhausted. Everything on screen is now frozen.
    ///
    /// `at_local` is a pre-formatted local timestamp rather than an instant to
    /// format here: this module holds no clock and no formatter, and a frozen
    /// banner whose text is supplied is a banner a snapshot test can pin.
    Frozen {
        /// When the link was declared lost, already rendered for display.
        at_local: String,
    },
    /// One key.
    Key(KeyPress),
    /// The 1s heartbeat. `now` is what advances every running sheep's uptime.
    Tick {
        /// The current instant, read by the caller.
        now: Instant,
    },
    /// The terminal changed size; nothing to update but the frame is stale.
    Resize,
    /// One reading of the machine this lookout is running on, off the 1-second
    /// heartbeat. `None` means `sysinfo` does not support this platform —
    /// which is a real, expected case, not a failure.
    ///
    /// Refused once the link is lost: see this arm in [`App::update`].
    Host {
        /// What the sampler saw, or `None` on an unsupported platform.
        sample: Option<super::source::HostSample>,
    },
    /// One refresh of the selected sheep's log files, in answer to an
    /// [`Effect::RefreshFeed`] this reducer asked for.
    ///
    /// Returns [`Effect::None`] unconditionally; see its arm.
    Bleats {
        /// What the read found, including what it could not show.
        tail: super::tail::Tail,
    },
    /// A request this dashboard asked for came back. `sent` is the echo tag.
    Replied {
        /// What was asked.
        sent: Sent,
        /// What the shepherd said, or why it could not be asked.
        result: Result<Response, RequestError>,
    },
    /// A request the caller could not hand to the link task.
    ///
    /// The reducer has already entered the in-flight state by the time
    /// `run_ui` tries to send, so a `try_send` that fails has to come back:
    /// otherwise the bar keeps saying "sent, waiting for the shepherd" about a
    /// request nobody has, and the one-action-at-a-time guard refuses every
    /// later action for the life of the process.
    Unsent {
        /// What could not be sent.
        sent: Sent,
    },
}

/// What the caller has to do after an update.
///
/// Not `Copy`: [`Self::Send`] carries a [`Sent`], whose `Sent::Action`
/// variant carries a `String`. Nothing matches on an `Effect` by reference,
/// so no call site changes for it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Effect {
    /// Nothing.
    None,
    /// Ask the link task for a `ListFlock` now, rather than at the next tick.
    PollNow,
    /// Re-read the selected sheep's log files and hand the result back as
    /// [`Msg::Bleats`].
    ///
    /// The feed has no timer of its own. It rides this: a snapshot produces
    /// one (so the two-second listing, a drop repair, a lag repair and `r`
    /// all refresh it), and so does a selection that actually moved. See the
    /// phase plan's design decision 2.
    RefreshFeed,
    /// Re-read the selected sheep's log files AND ask the shepherd for its
    /// lambs.
    ///
    /// Held apart from [`Self::RefreshFeed`] because the two triggers differ:
    /// a snapshot must refresh the feed, since the selected row's log paths
    /// can have changed, and must not fetch lambs, since it fires every two
    /// seconds. Returned by `select_at` when the selection actually moved, and
    /// by the callers of `reseat` on the same condition they already use to
    /// choose between `RefreshFeed` and `None`.
    RefreshSelected,
    /// Send a request to the shepherd. Raised by [`App::confirm`] once an
    /// armed action's Enter lands; `super::run_ui` is what actually sends it.
    Send(Sent),
    /// Leave.
    Quit,
}

/// The connection's state, as the dashboard reports it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Link {
    /// Connected and subscribed.
    Live,
    /// Re-dialling. `attempt` is 1-based and bounded by
    /// `super::link::RECONNECT_ATTEMPTS`.
    Retrying {
        /// Which attempt is in flight.
        attempt: u32,
    },
    /// The ladder is exhausted. Terminal: nothing moves this state, and the
    /// values on screen stay exactly as they were.
    Lost {
        /// When it was declared lost, already rendered for display.
        at_local: String,
    },
}

/// One sheep's row: what the shepherd said, and when it said it.
#[derive(Debug, Clone)]
pub struct Row {
    /// The shepherd's own snapshot of this sheep.
    pub info: ProcessInfo,
    /// When [`Self::info`] was received — the origin for this row's live
    /// uptime, so a value two seconds old is never rendered as current.
    pub anchor: Instant,
}

/// One request the dashboard asked the link task to send, carried back on the
/// reply so it can be routed.
///
/// An echo tag rather than a correlation id: the answer to a request can be an
/// `Err` that carries no shape of its own, so the only thing that reliably
/// says which request a reply belongs to is the request itself, handed along
/// beside it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Sent {
    /// The selected sheep's process tree.
    Lambs {
        /// Which sheep was asked about.
        id: u32,
    },
    /// One action against one sheep. `name` rides along so a reply can be
    /// reported by name even after the sheep has left the flock.
    Action {
        /// Which verb.
        verb: ActionVerb,
        /// The pinned sheep.
        id: u32,
        /// Its name at arm time.
        name: String,
    },
}

impl Sent {
    /// The wire request this asks for.
    #[must_use]
    pub fn request(&self) -> Request {
        match self {
            Self::Lambs { id } => Request::Describe {
                selector: SelectorSpec::Id(*id),
            },
            Self::Action { verb, id, .. } => {
                let selector = SelectorSpec::Id(*id);
                match verb {
                    ActionVerb::Stop => Request::Stop { selector },
                    ActionVerb::Restart => Request::Restart { selector },
                    ActionVerb::Reload => Request::Reload { selector },
                }
            }
        }
    }
}

/// What one lamb fetch came back with.
///
/// Three variants because `ProcessInfo::lambs` distinguishes three states and
/// the pane says three different sentences. The CLI has wording for only one
/// of them: `output::emit_described` skips the lamb caption for `None` and for
/// an empty vector alike, so there is nothing to borrow for the other two and
/// the pane says its own.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LambWalk {
    /// The shepherd walked the process table. Possibly to no descendants.
    Walked(Vec<Lamb>),
    /// The reply carried no walk at all, which for a `Describe` means this
    /// sheep has no pid to walk from.
    NotWalked,
    /// The request did not come back, or came back as something this binary
    /// does not understand.
    Failed,
}

/// One lamb reading, and which sheep it was taken for.
#[derive(Debug, Clone)]
pub struct LambReading {
    id: u32,
    at: Instant,
    walk: LambWalk,
}

/// A short line the status bar shows instead of the key hints, cleared by the
/// next keypress.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Notice {
    text: String,
    /// True for a refusal or a damage report — the status bar picks
    /// [`Palette::refusal`] over [`Palette::attention`].
    grave: bool,
}

impl Notice {
    /// Whether this notice is a refusal or a damage report rather than an
    /// informational one.
    #[must_use]
    pub fn is_grave(&self) -> bool {
        self.grave
    }
}

impl fmt::Display for Notice {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&self.text)
    }
}

/// What an action key does.
///
/// Three verbs, and deliberately not four: `start` is whistle's and the CLI's,
/// by Rin's ruling. Delete, scale, signal and whisper stay CLI-only for
/// whistle's own reasons: each takes a parameter a dashboard has nowhere to
/// put, or removes an app from the registry, which is the one action no
/// keypress should be one Enter away from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ActionVerb {
    /// `x`. Stops the sheep; it stays registered.
    Stop,
    /// `R`, on shift because `r` is refresh.
    Restart,
    /// `L`, on shift for symmetry with `R`.
    Reload,
}

impl ActionVerb {
    /// The word the prompt and every outcome sentence begin with.
    #[must_use]
    pub const fn label(self) -> &'static str {
        match self {
            Self::Stop => "stop",
            Self::Restart => "restart",
            Self::Reload => "reload",
        }
    }
}

/// Whether an action has been sent yet.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Stage {
    /// Armed, waiting for the operator's Enter. Nothing has gone out.
    Armed,
    /// Sent, waiting for the shepherd.
    Sent,
}

/// The one action this dashboard is in the middle of.
///
/// The target is captured HERE, by id and by name, and never re-read from the
/// selection: a snapshot can land between the arming keypress and the Enter,
/// and a confirmation that re-read the cursor could act on a sheep the
/// operator never pointed at.
///
/// One field on [`App`] rather than two `Option`s, so "armed" and "in flight"
/// cannot both be true. That is the same claim the one-action-at-a-time rule
/// makes, made in the type instead of in a guard.
#[derive(Debug, Clone)]
struct Action {
    verb: ActionVerb,
    id: u32,
    name: String,
    /// When it was armed. Only an armed action expires.
    at: Instant,
    stage: Stage,
}

/// What the status bar needs to know about the action in progress.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ActionState<'a> {
    /// Which verb.
    pub verb: ActionVerb,
    /// The pinned sheep's id.
    pub id: u32,
    /// The pinned sheep's name, as it was when the key was pressed.
    pub name: &'a str,
    /// False while it is a question, true once it has gone out.
    pub sent: bool,
}

/// How long an armed confirm waits for its Enter.
///
/// Ten seconds. A prompt left armed while the operator walks away, followed by
/// an Enter typed at what they think is a shell, is the same fat finger
/// arriving by a slower route. It rides `Msg::Tick`, so it costs one `Instant`
/// comparison and no timer (A11).
pub const CONFIRM_EXPIRY: Duration = Duration::from_secs(10);

/// The sentence `r` gives when the link is gone. The action keys refuse with
/// the same one, so the two cannot drift apart.
const LINK_GONE: &str = "the shepherd is gone — nothing left to ask";

/// The whole dashboard's state.
#[derive(Debug)]
pub struct App {
    flock: BTreeMap<u32, Row>,
    /// Which sheep the detail pane and the bleats feed describe.
    ///
    /// An **id**, not an index. The flock map is replaced wholesale every two
    /// seconds, so an index survives a `shep delete` of an earlier row by
    /// silently pointing at a different sheep — and every pane below the table
    /// would then describe that different sheep with nothing on screen
    /// changing. `None` only for an empty flock.
    ///
    /// The viewport offset is derived from this rather than stored beside it
    /// ([`super::view::flock::scroll_offset`]), which is what makes a
    /// disagreement between a stored offset and a stored cursor impossible
    /// rather than merely unlikely.
    selected: Option<u32>,
    /// The live substring filter over sheep NAMES, empty when there is none.
    ///
    /// Case-insensitive `contains`, and nothing else: not the CLI's selector
    /// grammar, not a regex, no understanding of `fold:`, `all` or ids. The
    /// grammar is exact-match on both the variants an operator would type
    /// while narrowing, so it cannot narrow as you type, and a half-typed
    /// `/re` parses as a search for a sheep literally named `/re` rather than
    /// refusing. See the design's feature 1 and assumption A1.
    ///
    /// Taken literally, spaces included, with no trimming (A6): this repo does
    /// not widen an accepted input format without a basis in the spec.
    filter: String,
    /// Which keymap [`super::input::map_key`] is called with. Normal until
    /// `/` opens the box; the reducer, not the keymap, owns this state.
    mode: InputMode,
    link: Link,
    notice: Option<Notice>,
    palette: Palette,
    control: Control,
    /// The `$SHEP_HOME` this lookout watches, for the title line. Held here
    /// rather than threaded through [`super::view::draw`]: it never changes for
    /// the life of the process, and a render function taking it as an argument
    /// would make every call site — including eight scene fixtures — carry it.
    home: String,
    /// The clock the view reads. Advanced by [`Msg::Tick`] — and deliberately
    /// NOT advanced once the link is [`Link::Lost`], which is what stops a
    /// frozen dashboard's uptime column from counting up for a sheep nothing
    /// can see.
    now: Instant,
    /// The last host reading, or `None` before the first heartbeat and on a
    /// platform `sysinfo` does not support. [`Self::host_unsupported`] tells
    /// the strip which of the two it is looking at.
    host: Option<super::source::HostSample>,
    /// True once a sample has come back `None` — the two `None`s mean
    /// different things and the strip says different sentences for them.
    host_unsupported: bool,
    /// The selected sheep's most recent output, as of the last refresh.
    /// [`Tail::default`](super::tail::Tail::default) before the first one —
    /// an empty, unlabelled tail, which
    /// [`super::view::bleats::feed_lines`] reads the same way it reads a
    /// sheep that has genuinely written nothing.
    feed: super::tail::Tail,
    /// The last lamb reading, or `None` before there has been one.
    ///
    /// Keyed by the id it was taken for, so a reading for a sheep that is no
    /// longer selected, and a request a full channel dropped, both read as
    /// "not read yet" without a second field tracking them.
    lambs: Option<LambReading>,
    /// The one action this dashboard is in the middle of, or `None`.
    action: Option<Action>,
}

impl App {
    /// A dashboard with an empty flock, a live link, and no notice.
    #[must_use]
    pub fn new(palette: Palette, control: Control, home: String, now: Instant) -> Self {
        Self {
            flock: BTreeMap::new(),
            selected: None,
            filter: String::new(),
            mode: InputMode::Normal,
            link: Link::Live,
            notice: None,
            palette,
            control,
            home,
            now,
            host: None,
            host_unsupported: false,
            feed: super::tail::Tail::default(),
            lambs: None,
            action: None,
        }
    }

    /// Applies one message and reports what the caller must do next.
    pub fn update(&mut self, msg: Msg) -> Effect {
        match msg {
            Msg::Snapshot { rows, at } => {
                // A snapshot cannot arrive after a freeze — the link task has
                // ended by then, so there is nothing left to produce one. If
                // one does, it is a message from a task that should not exist,
                // and accepting it would silently un-freeze a dashboard whose
                // banner says otherwise.
                if matches!(self.link, Link::Lost { .. }) {
                    return Effect::None;
                }
                let previous = self.selected_index();
                self.flock = rows
                    .into_iter()
                    .map(|info| (info.id, Row { info, anchor: at }))
                    .collect();
                self.reseat(previous);
                self.forget_missing_target();
                // Unconditional, and NOT `if reseat(..)`: the paths on the
                // selected row may have changed even when the selection did
                // not, and this is the whole of the feed's cadence. See design
                // decision 2.
                Effect::RefreshFeed
            }
            Msg::Event(event) => self.on_event(event),
            Msg::BusLagged { count } => {
                self.notice = Some(Notice {
                    text: format!(
                        "lookout fell behind and lost {count} events; re-reading the flock"
                    ),
                    grave: false,
                });
                Effect::PollNow
            }
            Msg::Retrying { attempt } => {
                if !matches!(self.link, Link::Lost { .. }) {
                    self.link = Link::Retrying { attempt };
                    self.disarm_on_link_change();
                }
                Effect::None
            }
            Msg::Relinked => {
                if !matches!(self.link, Link::Lost { .. }) {
                    self.link = Link::Live;
                }
                Effect::None
            }
            Msg::Frozen { at_local } => {
                self.link = Link::Lost { at_local };
                self.disarm_on_link_change();
                Effect::None
            }
            Msg::Tick { now } => {
                // The one line that keeps a frozen dashboard honest.
                if !matches!(self.link, Link::Lost { .. }) {
                    self.now = now;
                    let expired = self.action.as_ref().is_some_and(|action| {
                        action.stage == Stage::Armed
                            && now.saturating_duration_since(action.at) >= CONFIRM_EXPIRY
                    });
                    if expired {
                        self.action = None;
                    }
                }
                Effect::None
            }
            Msg::Resize => Effect::None,
            Msg::Key(key) => self.on_key(key),
            Msg::Host { sample } => {
                // The one line that keeps a frozen dashboard honest, for the
                // second time. `Msg::Tick` stops advancing `now` once the link
                // is lost; this stops the strip for the same reason. A single
                // line ticking over on a screen whose banner says the values
                // are frozen is a contradiction on one frame.
                if matches!(self.link, Link::Lost { .. }) {
                    return Effect::None;
                }
                self.host_unsupported = sample.is_none();
                self.host = sample;
                Effect::None
            }
            // Always `Effect::None`. A reducer that answered its own feed
            // update with another refresh request would spin the UI task at
            // full tilt; the `let _ =` at the call site in `run_ui` is
            // deliberate rather than lazy.
            //
            // The `Link::Lost` guard is load-bearing, not decorative: `run_ui`
            // arms its coalesced read from a `feed_dirty` flag set BEFORE a
            // freeze can land, so a read requested a moment earlier can still
            // be in flight when `Msg::Frozen` arrives and this arm sees the
            // stale read after it. Without the guard that read reaches the
            // rendered frame under a banner saying the values are frozen —
            // the same contradiction-on-one-frame design decision 7 already
            // refuses for the host strip and the uptime clock.
            Msg::Bleats { tail } => {
                if matches!(self.link, Link::Lost { .. }) {
                    return Effect::None;
                }
                self.feed = tail;
                Effect::None
            }
            Msg::Replied { sent, result } => match sent {
                Sent::Lambs { id } => self.on_lambs(id, result),
                Sent::Action { verb, id, name } => self.on_action_reply(verb, id, &name, result),
            },
            Msg::Unsent { sent } => match sent {
                Sent::Action { verb, id, name } => {
                    self.action = None;
                    self.notice = Some(Notice {
                        // "it was not sent", and no cause. The reducer does
                        // not know one: the channel holds 2, it is shared with
                        // lamb fetches, and `run_connected` awaits each
                        // request inline, so `Full` is reachable while the
                        // shepherd is perfectly reachable and merely slow.
                        // Naming a cause nothing observed is the failure the
                        // `-` CPU cell exists to prevent.
                        text: format!("{} {name} (id {id}): it was not sent", verb.label()),
                        grave: true,
                    });
                    Effect::None
                }
                // A dropped lamb fetch already reads as "not read yet", which
                // is what the pane says. Nothing to report and nothing to
                // clear.
                Sent::Lambs { .. } => Effect::None,
            },
        }
    }

    /// Records one lamb reading. Always [`Effect::None`]: a reducer that
    /// answered a reading with another request would spin the UI task.
    fn on_lambs(&mut self, id: u32, result: Result<Response, RequestError>) -> Effect {
        // The same guard `Msg::Bleats` carries, for the same reason: this
        // fetch is armed before a freeze can land, and a reading that reached
        // the frame afterwards would be newer than the banner saying the
        // values are frozen.
        if matches!(self.link, Link::Lost { .. }) {
            return Effect::None;
        }
        let walk = match result {
            Ok(Response::Described(rows)) => rows
                .into_iter()
                .find(|info| info.id == id)
                .map_or(LambWalk::Failed, |info| {
                    info.lambs.map_or(LambWalk::NotWalked, LambWalk::Walked)
                }),
            // An `Err`, or a reply this binary does not recognise. Neither is
            // an empty walk, and reporting either as one would say "none
            // found" about a process table nobody read.
            _ => LambWalk::Failed,
        };
        self.lambs = Some(LambReading {
            id,
            at: self.now,
            walk,
        });
        Effect::None
    }

    /// One action's answer: the shepherd's rows upserted, and one sentence.
    ///
    /// No provisional row state is invented anywhere on this path. The three
    /// replies carry `Vec<ProcessInfo>`, so the table updates from the
    /// shepherd's own words; an `online, restart sent...` in the STATUS column
    /// would be a guess printed in the one column whose whole job is to be
    /// true, and it would have to negotiate with the narrow-terminal column
    /// drop order for the privilege.
    fn on_action_reply(
        &mut self,
        verb: ActionVerb,
        id: u32,
        name: &str,
        result: Result<Response, RequestError>,
    ) -> Effect {
        self.action = None;
        let prefix = format!("{} {name} (id {id})", verb.label());
        // Each verb accepts its own reply and no other. A `Stopped` answering
        // a `Restart` carries rows and would upsert perfectly happily, which
        // is why the guards are on the arms rather than a single
        // rows-carrying match.
        let rows = match result {
            Ok(Response::Stopped(rows)) if verb == ActionVerb::Stop => rows,
            Ok(Response::Restarted(rows)) if verb == ActionVerb::Restart => rows,
            Ok(Response::Reloading(rows)) if verb == ActionVerb::Reload => rows,
            Ok(_unrecognised) => {
                self.notice = Some(Notice {
                    text: format!(
                        "{prefix}: the shepherd answered something this lookout does not understand"
                    ),
                    grave: true,
                });
                return Effect::None;
            }
            // The daemon's own message, not `RequestError`'s full `Display`:
            // the latter interpolates the code with `{:?}` and produces
            // "the daemon reported NotFound: ...", which puts a Rust
            // identifier on an operator's screen. The message alone is
            // already a sentence a human wrote.
            Err(RequestError::Rpc(err)) => {
                self.notice = Some(Notice {
                    text: format!("{prefix}: {}", err.message),
                    grave: true,
                });
                return Effect::None;
            }
            Err(other) => {
                self.notice = Some(Notice {
                    text: format!("{prefix}: {other}"),
                    grave: true,
                });
                return Effect::None;
            }
        };
        let anchor = self.now;
        let was_empty = self.flock.is_empty();
        for info in rows {
            self.flock.insert(info.id, Row { info, anchor });
        }
        self.notice = Some(Notice {
            text: format!("{prefix}: {}", outcome(verb)),
            grave: false,
        });
        if was_empty && self.reseat(None) {
            return Effect::RefreshSelected;
        }
        Effect::None
    }

    fn on_event(&mut self, event: BusEvent) -> Effect {
        match event {
            BusEvent::Process { event, info, .. } => {
                if matches!(event, ProcessEventKind::Delete) {
                    let previous = self.selected_index();
                    self.flock.remove(&info.id);
                    self.forget_missing_target();
                    return if self.reseat(previous) {
                        Effect::RefreshSelected
                    } else {
                        Effect::None
                    };
                }
                // An upsert cannot orphan the selection from `self.flock` —
                // the row it names either already existed (so the id is
                // still in the map) or is new (so it cannot be the one the
                // selection pointed at) — but it CAN orphan it from the
                // VISIBLE sequence `reseat` and `j`/`k` actually walk: a
                // rename that moves the selected row out of the current
                // filter leaves its id in `self.flock` while dropping it out
                // of `visible_ids()`. The old `was_empty`-only guard missed
                // that case (Phase 16 review Minor #6): the cursor stayed
                // pinned to an id the table had stopped drawing, and `j`/`k`
                // did nothing until the next snapshot repaired it. Reading
                // `previous` before the insert and always reseating covers
                // both that case and the empty-to-non-empty one below with
                // the same call — `reseat` itself is a no-op read when the
                // selection is still seated, so the common case costs one
                // cheap check.
                let previous = self.selected_index();
                let anchor = self.now;
                self.flock.insert(info.id, Row { info, anchor });
                if self.reseat(previous) {
                    return Effect::RefreshSelected;
                }
                Effect::None
            }
            // The shepherd's own outbound queue overflowed for this
            // subscriber. Deliberately worded differently from
            // `Msg::BusLagged` above: the two failures live on opposite ends of
            // the connection, and an operator cannot tell which end to
            // investigate if they read the same. `bleats` pins the identical
            // distinction.
            BusEvent::Dropped { count } => {
                self.notice = Some(Notice {
                    text: format!("the shepherd dropped {count} events; re-reading the flock"),
                    grave: false,
                });
                Effect::PollNow
            }
            // A notice, not an exit. `bleats` prints this and then leaves,
            // which is right for a follow; a standing dashboard that vanished
            // when the shepherd went down would take the last known state with
            // it. The link task will find the socket gone, climb the reconnect
            // ladder, and freeze if it runs out.
            BusEvent::DaemonShutdown => {
                self.notice = Some(Notice {
                    text: "the shepherd is shutting down".to_string(),
                    grave: true,
                });
                Effect::None
            }
            // `BusEvent` is `#[non_exhaustive]`: a variant a newer shepherd
            // added must not take the dashboard down, and must not be reported
            // as anything. `Dropped` and `DaemonShutdown` are NOT in this arm
            // — both are named variants this binary understands.
            _ => Effect::None,
        }
    }

    fn on_key(&mut self, key: KeyPress) -> Effect {
        // While the box is open every KEY is text. That is not the same as
        // nothing being able to raise a notice: three of them arrive on
        // messages that are not keys at all, and the status bar rather than
        // this branch is what keeps them off the query. See `on_text_key`.
        if self.mode == InputMode::Text {
            return self.on_text_key(key);
        }
        // Checked BEFORE the ordinary dispatch, and a cancelling keypress is
        // CONSUMED. A stray `j` during a pending confirm cancels it and does
        // not also move the selection: if it did both, the operator would see
        // the prompt vanish and the cursor move, and the next reflexive Enter
        // would act on a target they had already lost track of.
        //
        // Cancelling is silent. Nothing happened, and reporting nothing as if
        // it were something trains an operator to ignore the status bar.
        //
        // One consequence worth naming: `/` cancels before it opens the filter
        // box, so a confirm and a filter edit can never coexist. That closes
        // the whole interaction between the two features by construction
        // rather than by rule.
        if self
            .action
            .as_ref()
            .is_some_and(|action| action.stage == Stage::Armed)
        {
            if key == KeyPress::Confirm {
                return self.confirm();
            }
            // The one key the cancel does not consume. `input.rs`'s own doc
            // says why: dropping Ctrl-C would leave the most reflexive way out
            // of a terminal program doing nothing, and the operator's next
            // move is `kill -9` from another window, past every restore path
            // `super::term` has. Quitting DISCARDS the confirm rather than
            // acting on it, so the property this rule exists for is untouched;
            // that property is about a cancelling key also doing its ordinary
            // job on a target the operator has lost track of. Text mode makes
            // the same carve-out. See the phase plan's "Shapes the design
            // named" #4.
            if key == KeyPress::Quit {
                return Effect::Quit;
            }
            self.action = None;
            return Effect::None;
        }
        self.notice = None;
        match key {
            KeyPress::Quit => Effect::Quit,
            // The one key whose meaning depends on state, and the screen says
            // which meaning is in force: the bar reads `esc clear` for
            // exactly as long as clearing is what it does.
            KeyPress::Escape => {
                if self.filter.is_empty() {
                    Effect::Quit
                } else {
                    self.set_filter(String::new())
                }
            }
            // The one key that does I/O, and it refuses honestly once the
            // link task has ended: its poll receiver is gone by then, so an
            // `Effect::PollNow` would be a `try_send` into a closed channel
            // and the operator would get silence with no reason for it.
            KeyPress::Refresh => {
                if matches!(self.link, Link::Lost { .. }) {
                    self.notice = Some(Notice {
                        text: LINK_GONE.to_string(),
                        grave: true,
                    });
                    return Effect::None;
                }
                Effect::PollNow
            }
            KeyPress::SelectUp => self.select_by(-1),
            KeyPress::SelectDown => self.select_by(1),
            KeyPress::SelectFirst => self.select_at(0),
            KeyPress::SelectLast => self.select_at(self.visible_len().saturating_sub(1)),
            KeyPress::Action(verb) => self.arm(verb),
            // Enter means nothing outside an armed confirm. It reaches this
            // match whenever nothing is armed, which includes while an action
            // is IN FLIGHT, because the routing rule above only fires on
            // `Stage::Armed`. Named rather than swept into a wildcard: on the
            // one key whose job is to confirm a stop, an unspecified arm is
            // one edit away from being the wrong arm.
            KeyPress::Confirm => Effect::None,
            KeyPress::FilterStart => {
                self.mode = InputMode::Text;
                Effect::None
            }
            // `map_key` produces these only in text mode, which the branch at
            // the top of this function has already taken. Named rather than
            // wildcarded so a future variant does not fall silently into an
            // arm that ignores it.
            KeyPress::FilterChar(_)
            | KeyPress::FilterBackspace
            | KeyPress::FilterApply
            | KeyPress::FilterAbandon => Effect::None,
        }
    }

    /// Arms a confirm, or refuses and says why.
    ///
    /// Every refusal happens HERE rather than at confirm time, so an operator
    /// never answers a question that was never going to be honoured. Every
    /// sentence is literal: the standing rule is that nothing about damage
    /// gets charming, and a stop is damage.
    ///
    /// The ladder's order is the design's error table, read top to bottom:
    /// gate, link, nothing selected, one already in flight. It only shows when
    /// two conditions hold at once, and there is no reason to reorder an
    /// approved table.
    fn arm(&mut self, verb: ActionVerb) -> Effect {
        let refusal = if self.control == Control::ReadOnly {
            Some("read-only: actions need --allow-control".to_string())
        } else if let Link::Retrying { attempt } = self.link {
            // NOT `LINK_GONE` here: that sentence says the shepherd is gone,
            // which is false while it is still being redialled, and a
            // refusal saying so under a banner that says "reconnecting" was
            // Phase 16 review Minor #8 — two contradictory claims on one
            // frame. This is the status bar's own sentence for the state
            // (`view/status.rs`), so the refusal agrees with the banner
            // above it instead of overriding it.
            Some(format!(
                "the shepherd stopped answering — reconnecting (attempt {attempt})"
            ))
        } else if matches!(self.link, Link::Lost { .. }) {
            // The same sentence `r` gives once the ladder is exhausted: at
            // that point the shepherd really is gone, so `LINK_GONE` is
            // literally true rather than merely reused for convenience.
            Some(LINK_GONE.to_string())
        } else if self.selected_row().is_none() {
            Some("no sheep is selected".to_string())
        } else if self.action.is_some() {
            Some("one action is already in flight".to_string())
        } else {
            None
        };
        if let Some(text) = refusal {
            self.notice = Some(Notice { text, grave: true });
            return Effect::None;
        }
        let row = self.selected_row().expect("checked just above");
        self.action = Some(Action {
            verb,
            id: row.info.id,
            name: row.info.name.clone(),
            at: self.now,
            stage: Stage::Armed,
        });
        Effect::None
    }

    /// The operator's Enter. Sends, or refuses because the target left.
    fn confirm(&mut self) -> Effect {
        let Some(action) = self.action.take() else {
            return Effect::None;
        };
        // The whole flock, not the visible set: a filter typed after arming
        // hides a sheep, it does not remove it.
        if !self.flock.contains_key(&action.id) {
            self.notice = Some(Notice {
                text: format!(
                    "{} {} (id {}): it is no longer in the flock",
                    action.verb.label(),
                    action.name,
                    action.id
                ),
                grave: true,
            });
            return Effect::None;
        }
        let sent = Sent::Action {
            verb: action.verb,
            id: action.id,
            name: action.name.clone(),
        };
        self.action = Some(Action {
            stage: Stage::Sent,
            ..action
        });
        Effect::Send(sent)
    }

    /// Takes an armed prompt off the screen once its sheep is gone, rather
    /// than leaving a question about nothing. Called from the `Snapshot` arm
    /// and from the `Delete` arm; an action already in flight keeps its line.
    fn forget_missing_target(&mut self) {
        let gone = self.action.as_ref().is_some_and(|action| {
            action.stage == Stage::Armed && !self.flock.contains_key(&action.id)
        });
        if gone {
            self.action = None;
        }
    }

    /// Takes an armed prompt off the screen when the link stops being live.
    ///
    /// Called from the `Msg::Retrying` and `Msg::Frozen` arms. A9 refuses to
    /// ARM unless the link is `Live`; without this, a prompt armed a moment
    /// earlier would outlive the connection it was going to be sent over, and
    /// on a frozen dashboard it would never expire either, because `now` stops
    /// advancing and the expiry check rides it. Silent, like every other
    /// cancel: nothing happened, and the banner appearing on the same frame
    /// already says why.
    ///
    /// An action already SENT keeps its line. It is a real request, and
    /// `run_connected` answers it with an `Err` before its loop ends, so the
    /// in-flight line always resolves rather than hanging.
    fn disarm_on_link_change(&mut self) {
        if self
            .action
            .as_ref()
            .is_some_and(|action| action.stage == Stage::Armed)
        {
            self.action = None;
        }
    }

    /// The filter box's keymap.
    ///
    /// Ctrl-C still quits: in raw mode it is a key event and not a signal,
    /// and a text box that swallowed it would leave the operator reaching for
    /// `kill -9` from another window, past every restore path
    /// [`super::term`] has.
    ///
    /// Deliberately does NOT clear [`Self::notice`], unlike the normal-mode
    /// branch above it. A notice can be raised while this box is open, by
    /// `Msg::BusLagged`, `BusEvent::Dropped` or `BusEvent::DaemonShutdown`,
    /// none of which is a keypress; clearing here would destroy it because
    /// somebody was mid-word. The status bar hides it under the box instead
    /// and shows it when the box closes. See the phase plan's "Shapes the
    /// design named" #2.
    fn on_text_key(&mut self, key: KeyPress) -> Effect {
        match key {
            KeyPress::Quit => Effect::Quit,
            KeyPress::FilterChar(typed) => {
                let mut query = self.filter.clone();
                query.push(typed);
                self.set_filter(query)
            }
            KeyPress::FilterBackspace => {
                let mut query = self.filter.clone();
                query.pop();
                self.set_filter(query)
            }
            KeyPress::FilterApply => {
                self.mode = InputMode::Normal;
                Effect::None
            }
            KeyPress::FilterAbandon => {
                self.mode = InputMode::Normal;
                self.set_filter(String::new())
            }
            _ => Effect::None,
        }
    }

    /// The ids the table draws, in id order: the whole flock, or whatever the
    /// filter leaves of it.
    ///
    /// [`Self::rows`], [`Self::select_at`], [`Self::select_by`],
    /// [`Self::reseat`] and [`Self::selected_index`] all read this sequence
    /// and nothing else. That is the whole point of it: a filter that hid rows
    /// `j` and `k` still stepped over would move the cursor onto a sheep
    /// nobody can see, and every pane below the table would then describe that
    /// sheep with nothing on screen saying so.
    ///
    /// One lowercase allocation per call, for the query only. The flock is a
    /// `BTreeMap` an operator is looking at, so it is tens of rows, not
    /// thousands, and a cached needle would be a second source of truth for
    /// [`Self::filter`].
    fn visible_ids(&self) -> impl Iterator<Item = u32> + '_ {
        let needle = self.filter.to_lowercase();
        self.flock.iter().filter_map(move |(id, row)| {
            let shown = needle.is_empty() || row.info.name.to_lowercase().contains(&needle);
            shown.then_some(*id)
        })
    }

    /// How many rows the table draws.
    fn visible_len(&self) -> usize {
        self.visible_ids().count()
    }

    /// Puts the selection back on a real sheep after the flock changed.
    ///
    /// `previous_index` is where the selection sat **before** the change, read
    /// while the old map was still in place. A selection whose id survived is
    /// left alone, whatever row it now occupies. One that did not falls to
    /// whatever occupies the same position, clamped to the last row — not to
    /// row 0, which would throw an operator back to the top of a
    /// two-hundred-sheep flock every time an unrelated sheep was deleted.
    ///
    /// Returns whether the selection changed, which is what decides between
    /// [`Effect::RefreshSelected`] and [`Effect::None`].
    fn reseat(&mut self, previous_index: Option<usize>) -> bool {
        // `selected_index`, NOT `flock.contains_key`: a selection the filter
        // is hiding is not seated, however present its id still is in the map.
        // This one line is the difference between a cursor that follows the
        // filter and one that wanders behind it.
        //
        // It comes FIRST, where the shipped `flock.is_empty()` check used to.
        // The order is behaviour-preserving (a seated selection implies at
        // least one visible row), and it is what makes Step 1.6's mutation
        // able to reach the nothing-matches case: with the emptiness test
        // above it, reverting this line changes nothing when the query
        // matches no sheep, because the early return has already fired.
        if self.selected_index().is_some() {
            return false;
        }
        let before = self.selected;
        let visible = self.visible_len();
        if visible == 0 {
            self.selected = None;
            return before != self.selected;
        }
        let index = previous_index.unwrap_or(0).min(visible - 1);
        // Bound first, then assigned. `self.selected = self.visible_ids()
        // .nth(index);` does NOT compile: `visible_ids` returns
        // `impl Iterator<Item = u32> + '_`, whose temporary holds a shared
        // borrow of ALL of `self` to the end of the statement, so the
        // assignment overlaps it and rustc raises E0506. The line this
        // replaces borrowed one FIELD (`self.flock.keys()`), which is why it
        // was fine. `select_at` below takes the same two-line shape for the
        // same reason. Verified with rustc on an edition-2024 scaffold, both
        // ways round.
        let next = self.visible_ids().nth(index);
        self.selected = next;
        before != self.selected
    }

    /// Moves the selection by `delta` rows and reports whether it moved.
    ///
    /// Clamped rather than wrapping: wrapping a two-hundred-sheep flock from
    /// the last row to the first on one keypress loses the operator's place
    /// with nothing to undo it.
    fn select_by(&mut self, delta: isize) -> Effect {
        let Some(index) = self.selected_index() else {
            return Effect::None;
        };
        let next = index.saturating_add_signed(delta);
        self.select_at(next)
    }

    /// Selects the row at `index`, clamped to the flock, and reports whether
    /// that changed anything.
    ///
    /// `Effect::None` when it did not: [`Effect::RefreshSelected`] reads two
    /// files off disk and asks the shepherd for lambs, and a held `k` at the
    /// top of the flock must not do that once per keypress.
    fn select_at(&mut self, index: usize) -> Effect {
        let visible = self.visible_len();
        if visible == 0 {
            return Effect::None;
        }
        let next = self.visible_ids().nth(index.min(visible - 1));
        if next == self.selected {
            return Effect::None;
        }
        self.selected = next;
        // The cursor moved; whether anything is READ is a separate question.
        // A frozen dashboard re-reading live log files would put content on
        // screen newer than the banner saying the values are frozen, which is
        // the contradiction design decision 7 refuses for the host strip. The
        // detail pane still re-renders, from the frozen listing — that is data
        // already on the frame.
        if matches!(self.link, Link::Lost { .. }) {
            return Effect::None;
        }
        Effect::RefreshSelected
    }

    /// The flock the table draws, in id order: the whole flock, or whatever
    /// the filter leaves of it. See [`Self::all_rows`] for the unfiltered
    /// sequence a pane describing the machine as a whole, not the table's
    /// current view, needs instead.
    #[must_use]
    pub fn rows(&self) -> Vec<&Row> {
        self.visible_ids()
            .filter_map(|id| self.flock.get(&id))
            .collect()
    }

    /// Every sheep the shepherd last reported, in id order, whatever the
    /// filter hides.
    ///
    /// The host strip reads this rather than [`Self::rows`]: Phase 16 review
    /// Important #4 caught `flock cpu`/`flock mem` silently summing the FILTERED
    /// set while staying labelled `flock`, so a filter that matched nothing
    /// made the strip print `-` for a running flock the strip had simply
    /// stopped looking at. `-` is reserved for a genuine unknown reading
    /// ([`ProcessInfo::cpu_percent`]'s own doc), not for "the table's query
    /// happens to exclude everything right now", and the title bar already
    /// carries the filtered-vs-total distinction (`2 of 6 in the flock`) so
    /// the strip does not have to.
    #[must_use]
    pub fn all_rows(&self) -> Vec<&Row> {
        self.flock.values().collect()
    }

    /// Replaces the filter and puts the selection back on a visible sheep.
    ///
    /// Private, and its only production caller is [`Self::on_key`]'s text-mode
    /// arm (Task 2). The reseat is the whole reason this is not a plain field
    /// assignment: a keystroke that narrows the query can hide the selected
    /// sheep, and the selection then falls to whatever occupies the same
    /// position, clamped to the last visible row, which is `reseat`'s shipped
    /// rule applied to a new cause.
    fn set_filter(&mut self, query: String) -> Effect {
        if self.filter == query {
            return Effect::None;
        }
        let previous = self.selected_index();
        self.filter = query;
        if self.reseat(previous) && !matches!(self.link, Link::Lost { .. }) {
            // The cursor moved, so the feed and the lambs are about to
            // describe a different sheep. A frozen dashboard reads nothing,
            // for the reason `select_at` already states.
            return Effect::RefreshSelected;
        }
        Effect::None
    }

    /// The filter as typed, empty when there is none.
    #[must_use]
    pub fn filter(&self) -> &str {
        &self.filter
    }

    /// Which keymap is currently in force, for `super::run_ui`'s call to
    /// [`super::input::map_key`].
    #[must_use]
    pub fn mode(&self) -> InputMode {
        self.mode
    }

    /// How many sheep the shepherd last reported, whatever the filter hides.
    ///
    /// The title reads this beside `rows().len()`. A title that could only
    /// read the narrowed count would understate the flock while a filter is
    /// on, which is the confident wrong number this dashboard's `-` cells and
    /// its frozen uptime clock both exist to avoid.
    #[must_use]
    pub fn flock_len(&self) -> usize {
        self.flock.len()
    }

    /// The selected sheep's id, or `None` for an empty flock.
    #[must_use]
    pub fn selected(&self) -> Option<u32> {
        self.selected
    }

    /// Which row of [`Self::rows`] the selection sits on.
    ///
    /// Derived every call rather than stored: see [`Self::selected`].
    #[must_use]
    pub fn selected_index(&self) -> Option<usize> {
        let id = self.selected?;
        self.visible_ids().position(|key| key == id)
    }

    /// The selected sheep's row, which the detail pane and the feed read.
    #[must_use]
    pub fn selected_row(&self) -> Option<&Row> {
        self.flock.get(&self.selected?)
    }

    /// The link state, as the status bar reports it.
    #[must_use]
    pub fn link(&self) -> &Link {
        &self.link
    }

    /// The current notice, if the last message left one.
    #[must_use]
    pub fn notice(&self) -> Option<&Notice> {
        self.notice.as_ref()
    }

    /// The resolved palette.
    #[must_use]
    pub fn palette(&self) -> Palette {
        self.palette
    }

    /// Whether actions are permitted.
    #[must_use]
    pub fn control(&self) -> Control {
        self.control
    }

    /// The `$SHEP_HOME` this lookout watches.
    #[must_use]
    pub fn home(&self) -> &str {
        &self.home
    }

    /// The last host reading, or `None` if there has not been one.
    #[must_use]
    pub fn host(&self) -> Option<super::source::HostSample> {
        self.host
    }

    /// Whether [`Self::host`] is `None` because the platform cannot be read,
    /// rather than because no heartbeat has fired yet.
    ///
    /// The strip says `usage is not available on this platform` for the first
    /// and `not read yet` for the second. They are different facts and an
    /// operator seeing the wrong one waits for numbers that are never coming.
    #[must_use]
    pub fn host_unsupported(&self) -> bool {
        self.host_unsupported
    }

    /// The selected sheep's most recent output, as of the last refresh.
    #[must_use]
    pub fn feed(&self) -> &super::tail::Tail {
        &self.feed
    }

    /// One sheep's uptime as of this dashboard's own clock, in milliseconds.
    ///
    /// A **running** sheep's uptime advances between polls, from the anchor its
    /// row carries — the alternative, showing a number that only moves every
    /// two seconds, reads as a frozen dashboard when it is not. A sheep that is
    /// not running does not advance: its `uptime_ms` is a historical fact about
    /// how long it ran, and animating it would invent one.
    ///
    /// While the link is [`Link::Lost`], nothing advances at all, because
    /// `self.now` stops.
    #[must_use]
    pub fn uptime_ms(&self, id: u32) -> Option<u64> {
        let row = self.flock.get(&id)?;
        if !matches!(row.info.status, ProcStatus::Online | ProcStatus::Starting) {
            return Some(row.info.uptime_ms);
        }
        let elapsed = self.now.saturating_duration_since(row.anchor);
        Some(row.info.uptime_ms.saturating_add(millis(elapsed)))
    }

    /// The lamb reading for sheep `id`, with its age in milliseconds as of
    /// this dashboard's own clock.
    ///
    /// `None` when there is no reading, or when the one there is was taken for
    /// a different sheep. The age comes from [`Self::now`], the same clock the
    /// uptime column reads, which means it stops when the dashboard freezes,
    /// for the same reason.
    #[must_use]
    pub fn lambs_for(&self, id: u32) -> Option<(&LambWalk, u64)> {
        let reading = self.lambs.as_ref().filter(|reading| reading.id == id)?;
        Some((
            &reading.walk,
            millis(self.now.saturating_duration_since(reading.at)),
        ))
    }

    /// The action in progress, for the status bar.
    #[must_use]
    pub fn action(&self) -> Option<ActionState<'_>> {
        let action = self.action.as_ref()?;
        Some(ActionState {
            verb: action.verb,
            id: action.id,
            name: &action.name,
            sent: action.stage == Stage::Sent,
        })
    }

    /// Overrides the control gate a fixture built with. `App::new` takes
    /// [`Control`] and every shipped fixture hard-codes [`Control::ReadOnly`];
    /// this is the one line that lets the action-key tests build a dashboard
    /// with the gate open without a second copy of `app_with`.
    #[cfg(test)]
    pub(crate) fn set_control_for_tests(&mut self, control: Control) {
        self.control = control;
    }

    /// Sets the filter directly, bypassing the reseat [`Self::set_filter`]
    /// does. `set_filter` is private to this module; this is the one line
    /// that lets `view::host`'s tests (a sibling module, not a descendant)
    /// build a dashboard with a filter already applied, the same shape
    /// [`Self::set_control_for_tests`] takes for the control gate.
    #[cfg(test)]
    pub(crate) fn set_filter_for_tests(&mut self, query: &str) {
        self.filter = query.to_string();
    }
}

/// Saturating `Duration` -> milliseconds. A lookout left open for 580 million
/// years is not the failure this guards; the cast is what clippy's
/// `cast_possible_truncation` would otherwise deny.
fn millis(d: Duration) -> u64 {
    u64::try_from(d.as_millis()).unwrap_or(u64::MAX)
}

/// What the bar says once the shepherd has answered.
///
/// Reload's wording is not decoration. `Response::Reloading` is an acceptance
/// and not a result, and the swaps arrive afterwards on the bus, which the
/// table already consumes.
const fn outcome(verb: ActionVerb) -> &'static str {
    match verb {
        ActionVerb::Stop => "the shepherd stopped it",
        ActionVerb::Restart => "the shepherd restarted it",
        ActionVerb::Reload => "accepted, the swaps report themselves as they happen",
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use shep_core::protocol::{ProcessEventKind, RpcError, RpcErrorCode};

    fn sheep(id: u32, name: &str, status: ProcStatus) -> ProcessInfo {
        ProcessInfo::builder(id, name, status)
            .pid(Some(1000 + id))
            .uptime_ms(60_000)
            .build()
    }

    fn started() -> (App, Instant) {
        let t0 = Instant::now();
        let mut app = App::new(
            Palette::detect(None, None, None),
            Control::ReadOnly,
            "/home/rin/.shep".to_string(),
            t0,
        );
        app.update(Msg::Snapshot {
            rows: vec![
                sheep(1, "web", ProcStatus::Online),
                sheep(2, "api", ProcStatus::Errored),
                sheep(3, "worker", ProcStatus::Online),
            ],
            at: t0,
        });
        (app, t0)
    }

    /// `started()`'s three sheep with the gate open and the cursor parked in
    /// the middle, on `api` at id 2.
    ///
    /// Mid-list on purpose. Half the tests below assert that a stray `j` did
    /// NOT move the cursor, and a cursor already clamped at either end would
    /// pass those tests whether the routing rule consumed the key or not.
    fn allowed() -> App {
        let t0 = Instant::now();
        let mut app = App::new(
            Palette::detect(None, None, None),
            Control::Allowed,
            "/home/rin/.shep".to_string(),
            t0,
        );
        app.update(Msg::Snapshot {
            rows: vec![
                sheep(1, "web", ProcStatus::Online),
                sheep(2, "api", ProcStatus::Online),
                sheep(3, "worker", ProcStatus::Online),
            ],
            at: t0,
        });
        app.update(Msg::Tick { now: t0 });
        app.update(Msg::Key(KeyPress::SelectDown));
        app
    }

    /// fails if an action key acts. It arms, and nothing has been sent: the
    /// whole point of the gate is that one keystroke in a dashboard somebody
    /// is reading does not become an action.
    #[test]
    fn an_action_key_arms_a_confirm_and_sends_nothing() {
        let mut app = allowed();
        assert_eq!(
            app.update(Msg::Key(KeyPress::Action(ActionVerb::Stop))),
            Effect::None
        );
        let armed = app.action().expect("armed");
        assert_eq!(armed.verb, ActionVerb::Stop);
        assert_eq!(armed.id, 2);
        assert_eq!(armed.name, "api");
        assert!(!armed.sent, "nothing has gone out");
    }

    /// fails if the action key itself confirms, or if only `Esc` cancels. A
    /// confirm the action key could complete is the double-tap the gate exists
    /// to catch, on a keyboard that may be repeating.
    #[test]
    fn only_enter_confirms_and_every_other_key_cancels() {
        for key in [
            KeyPress::SelectDown,
            KeyPress::SelectUp,
            KeyPress::SelectFirst,
            KeyPress::Refresh,
            KeyPress::Escape,
            KeyPress::FilterStart,
            KeyPress::Action(ActionVerb::Stop),
            KeyPress::Action(ActionVerb::Restart),
        ] {
            let mut app = allowed();
            app.update(Msg::Key(KeyPress::Action(ActionVerb::Stop)));
            assert!(app.action().is_some(), "armed before {key:?}");
            assert_eq!(
                app.update(Msg::Key(key)),
                Effect::None,
                "{key:?} sent something"
            );
            assert!(app.action().is_none(), "{key:?} did not cancel");
        }

        let mut app = allowed();
        app.update(Msg::Key(KeyPress::Action(ActionVerb::Stop)));
        assert!(
            matches!(app.update(Msg::Key(KeyPress::Confirm)), Effect::Send(_)),
            "and Enter is the one key that sends"
        );
    }

    /// fails if a cancelling keypress ALSO does its ordinary job. This is the
    /// failure mode the whole feature is about: the operator would see the
    /// prompt vanish and the cursor move, and the next reflexive Enter would
    /// act on a target they had already lost track of.
    ///
    /// Three assertions, because each catches a different half of the bug: the
    /// prompt is gone, the cursor did NOT move, and no effect leaked out. The
    /// selection is parked mid-list by `allowed()` so a `j` genuinely could
    /// move it.
    #[test]
    fn a_cancelling_key_is_consumed_and_does_not_also_move_the_selection() {
        let mut app = allowed();
        app.update(Msg::Key(KeyPress::Action(ActionVerb::Restart)));
        let before = app.selected();
        let effect = app.update(Msg::Key(KeyPress::SelectDown));
        assert!(app.action().is_none(), "the stray j cancelled the confirm");
        assert_eq!(app.selected(), before, "and did not also move the cursor");
        assert_eq!(effect, Effect::None, "nor ask for a feed read or a walk");
    }

    /// fails if the confirm re-reads the cursor at Enter time. A snapshot can
    /// land between the arming keypress and the Enter, and a confirmation
    /// built from `self.selected` would then act on a sheep the operator never
    /// pointed at.
    ///
    /// Task 7's original version of this test armed on id 2 then had a
    /// snapshot delete id 2's NEIGHBOUR, expecting that to move the cursor.
    /// It never did: `reseat`'s own rule is "an id that survived is left
    /// alone, whatever row it now occupies" — id 2 survived every one of
    /// those snapshots, so `self.selected` stayed 2 right alongside
    /// `action.id`, and the mutation this test exists to catch (reading
    /// `self.selected` instead of the pinned `action.id`) could not redden
    /// it. Phase 16 review Minor #7.
    ///
    /// This version applies a filter before arming, then has the snapshot
    /// RENAME the armed sheep out of that filter while a second sheep enters
    /// it. The armed id (2) still survives in `self.flock` — a rename is not
    /// a delete, so `confirm`'s "did the target leave" check must not fire —
    /// but it drops out of `visible_ids()`, so `reseat` moves the cursor to
    /// the other match. That genuinely separates `self.selected` (9) from
    /// `action.id` (2), which is what makes the pin observable: the fixed
    /// code sends id 2 under its arm-time name "api", and the mutation Task 7
    /// tried would instead send id 9 under whatever the snapshot named it.
    #[test]
    fn the_confirm_is_pinned_to_the_id_it_was_armed_on() {
        let mut app = allowed();
        app.set_filter("api".to_string());
        app.update(Msg::Key(KeyPress::Action(ActionVerb::Stop)));
        app.update(Msg::Snapshot {
            rows: vec![
                sheep(2, "gateway", ProcStatus::Online),
                sheep(9, "api-new", ProcStatus::Online),
            ],
            at: Instant::now(),
        });
        assert_eq!(
            app.selected(),
            Some(9),
            "sanity: the cursor followed the filter off the armed id"
        );
        let Effect::Send(sent) = app.update(Msg::Key(KeyPress::Confirm)) else {
            panic!("Enter sends");
        };
        assert_eq!(
            sent,
            Sent::Action {
                verb: ActionVerb::Stop,
                id: 2,
                name: "api".to_string()
            }
        );
    }

    /// fails if a confirm whose sheep left the flock sends anyway. The one
    /// refusal that has to happen at confirm time rather than at arm time.
    #[test]
    fn a_confirm_whose_sheep_left_the_flock_refuses_instead_of_sending() {
        let mut app = allowed();
        app.update(Msg::Key(KeyPress::Action(ActionVerb::Stop)));
        app.update(Msg::Event(BusEvent::Process {
            event: ProcessEventKind::Delete,
            info: sheep(2, "api", ProcStatus::Stopped),
            manually: true,
            at_ms: 0,
        }));
        // The prompt came off the screen as soon as the reducer learned the
        // sheep was gone, rather than sitting there as a question about
        // nothing.
        assert!(app.action().is_none());
        assert_eq!(app.update(Msg::Key(KeyPress::Confirm)), Effect::None);
    }

    /// fails if a prompt left armed while the operator walks away never
    /// expires. Driven by `Msg::Tick`, so there is no sleep here (IR-33).
    #[test]
    fn a_confirm_expires_after_ten_seconds_of_ticks() {
        let mut app = allowed();
        let t0 = Instant::now();
        app.update(Msg::Tick { now: t0 });
        app.update(Msg::Key(KeyPress::Action(ActionVerb::Reload)));
        app.update(Msg::Tick {
            now: t0 + Duration::from_secs(9),
        });
        assert!(app.action().is_some(), "nine seconds is still armed");
        app.update(Msg::Tick {
            now: t0 + Duration::from_secs(10),
        });
        assert!(app.action().is_none(), "ten is not");
    }

    /// fails if arming is allowed while the link is not live. An action typed
    /// during the eight second reconnect ladder would otherwise queue and land
    /// seconds later, on a connection the operator has stopped watching (A9).
    #[test]
    fn every_action_key_refuses_while_the_link_is_not_live() {
        for link in [
            Msg::Retrying { attempt: 2 },
            Msg::Frozen {
                at_local: "2026-08-16 09:00:00".to_string(),
            },
        ] {
            let mut app = allowed();
            app.update(link);
            app.update(Msg::Key(KeyPress::Action(ActionVerb::Stop)));
            assert!(app.action().is_none());
            assert!(app.notice().is_some_and(Notice::is_grave));
        }
    }

    /// fails if a second action can be armed while one is in flight. The
    /// in-flight line names one action; a second one would make it ambiguous
    /// which (A12).
    #[test]
    fn a_second_action_refuses_while_one_is_in_flight() {
        let mut app = allowed();
        app.update(Msg::Key(KeyPress::Action(ActionVerb::Stop)));
        app.update(Msg::Key(KeyPress::Confirm));
        assert!(app.action().is_some_and(|action| action.sent));
        app.update(Msg::Key(KeyPress::Action(ActionVerb::Restart)));
        assert_eq!(
            app.notice().map(ToString::to_string).as_deref(),
            Some("one action is already in flight")
        );
        let action = app.action().expect("the first one is untouched");
        assert_eq!(action.verb, ActionVerb::Stop);
        assert!(action.sent);
    }

    /// fails if the in-flight line is stored as a notice. A notice is cleared
    /// by the next keypress, and an in-flight action whose only sign on screen
    /// could be wiped by a stray `j` is a dashboard hiding something it knows.
    #[test]
    fn an_in_flight_line_survives_a_keypress() {
        let mut app = allowed();
        app.update(Msg::Key(KeyPress::Action(ActionVerb::Stop)));
        app.update(Msg::Key(KeyPress::Confirm));
        app.update(Msg::Key(KeyPress::SelectDown));
        assert!(
            app.action().is_some_and(|action| action.sent),
            "the keypress moved the cursor and left the in-flight state alone"
        );
    }

    /// fails if `q` or Ctrl-C stops quitting while a prompt is up. The routing
    /// rule consumes every other key; this is the one carve-out, and
    /// `input.rs`'s own doc says why: an operator whose most reflexive way out
    /// of a terminal program stops working reaches for `kill -9` from another
    /// window, past every restore path `term` has. Quitting discards the
    /// confirm rather than acting on it, so nothing this rule protects is
    /// weakened. See the phase plan's "Shapes the design named" #4.
    #[test]
    fn quit_still_quits_while_a_confirm_is_armed() {
        let mut app = allowed();
        app.update(Msg::Key(KeyPress::Action(ActionVerb::Stop)));
        assert_eq!(app.update(Msg::Key(KeyPress::Quit)), Effect::Quit);
    }

    /// fails if Enter does something when nothing is armed. It reaches the
    /// ordinary match in two states an operator can be in: nothing armed at
    /// all, and one action already in flight. Neither may send.
    #[test]
    fn enter_outside_an_armed_confirm_does_nothing() {
        let mut app = allowed();
        assert_eq!(app.update(Msg::Key(KeyPress::Confirm)), Effect::None);
        assert!(app.action().is_none());

        app.update(Msg::Key(KeyPress::Action(ActionVerb::Stop)));
        app.update(Msg::Key(KeyPress::Confirm));
        assert!(app.action().is_some_and(|action| action.sent), "in flight");
        assert_eq!(
            app.update(Msg::Key(KeyPress::Confirm)),
            Effect::None,
            "a second Enter does not re-send"
        );
    }

    /// fails if a request the channel could not take leaves the dashboard
    /// claiming an action is in flight forever, which would also refuse every
    /// later action for the life of the process.
    #[test]
    fn a_request_that_could_not_be_sent_says_so_and_clears_the_state() {
        let mut app = allowed();
        app.update(Msg::Key(KeyPress::Action(ActionVerb::Stop)));
        let Effect::Send(sent) = app.update(Msg::Key(KeyPress::Confirm)) else {
            panic!("Enter sends");
        };
        app.update(Msg::Unsent { sent });
        assert!(app.action().is_none());
        assert!(app.notice().is_some_and(Notice::is_grave));
    }

    /// fails if an armed prompt outlives the connection it was going to be
    /// sent over. Both halves matter: `Retrying` because A9 says an action
    /// typed during the reconnect ladder is refused rather than queued, and
    /// `Frozen` because a frozen dashboard's `now` stops advancing, so an
    /// armed prompt there would never expire either.
    #[test]
    fn a_link_that_stops_being_live_takes_an_armed_prompt_down() {
        for link in [
            Msg::Retrying { attempt: 2 },
            Msg::Frozen {
                at_local: "2026-08-16 09:00:00".to_string(),
            },
        ] {
            let mut app = allowed();
            app.update(Msg::Key(KeyPress::Action(ActionVerb::Stop)));
            assert!(app.action().is_some(), "armed while live");
            app.update(link);
            assert!(app.action().is_none(), "and gone once the link is not");
            assert_eq!(
                app.update(Msg::Key(KeyPress::Confirm)),
                Effect::None,
                "so Enter has nothing to send"
            );
        }
    }

    /// fails if the poll stops being the truth. A bus event upserts, but a
    /// snapshot REPLACES: the bus is lossy by construction, so a sheep the
    /// events invented — or one they never learned had been deleted — must not
    /// survive the next listing. Event-sourcing a lossy bus is the exact drift
    /// this reducer exists to prevent.
    #[test]
    fn a_snapshot_replaces_the_flock_wholesale() {
        let (mut app, t0) = started();
        app.update(Msg::Event(BusEvent::Process {
            event: ProcessEventKind::Start,
            info: sheep(9, "ghost", ProcStatus::Starting),
            manually: true,
            at_ms: 0,
        }));
        assert_eq!(app.rows().len(), 4, "the bus event upserted");

        app.update(Msg::Snapshot {
            rows: vec![sheep(1, "web", ProcStatus::Online)],
            at: t0,
        });
        assert_eq!(app.rows().len(), 1);
        assert!(app.rows().iter().all(|row| row.info.id == 1));
    }

    /// fails if a snapshot that shrinks the flock leaves the selection pointing
    /// at a row that no longer exists. The map is REPLACED wholesale every two
    /// seconds, so a selection that was valid for six sheep can outlive four of
    /// them — and this is the reseat rule, not the clamp `last_row` used to be.
    #[test]
    fn a_snapshot_that_shrinks_the_flock_pulls_the_selection_back() {
        let (mut app, t0) = started();
        app.update(Msg::Key(KeyPress::SelectLast));
        assert_eq!(app.selected_index(), Some(2));

        app.update(Msg::Snapshot {
            rows: vec![sheep(1, "web", ProcStatus::Online)],
            at: t0,
        });
        assert_eq!(
            app.selected_index(),
            Some(0),
            "the selection came back with the flock"
        );

        app.update(Msg::Snapshot {
            rows: vec![],
            at: t0,
        });
        assert_eq!(app.selected_index(), None, "an empty flock selects nothing");
    }

    /// fails if the selection stops being stored as an ID. The flock map is
    /// replaced wholesale every two seconds, so an INDEX cursor silently
    /// points at a different sheep the moment an earlier row is deleted —
    /// and the detail pane and the feed would then describe that different
    /// sheep with nothing on screen changing. This is the whole reason
    /// `selected` is a `u32` and not a `usize`.
    #[test]
    fn the_selection_follows_the_sheep_and_not_the_row_number() {
        let (mut app, t0) = started();
        app.update(Msg::Key(KeyPress::SelectDown));
        app.update(Msg::Key(KeyPress::SelectDown));
        assert_eq!(app.selected(), Some(3), "the third row, worker");

        // Sheep 1 goes away. `worker` is now row 1 rather than row 2 — an
        // index cursor would now be pointing at `api`.
        app.update(Msg::Snapshot {
            rows: vec![
                sheep(2, "api", ProcStatus::Errored),
                sheep(3, "worker", ProcStatus::Online),
            ],
            at: t0,
        });
        assert_eq!(app.selected(), Some(3), "still worker");
        assert_eq!(app.selected_index(), Some(1), "which is now row 1");
    }

    /// fails if a selection whose sheep was deleted jumps to the top of the
    /// flock. It falls to whatever now occupies the same POSITION, clamped —
    /// throwing an operator back to row 0 of a two-hundred-sheep flock every
    /// time an unrelated sheep is deleted is the behaviour this rejects.
    #[test]
    fn a_deleted_selection_falls_to_the_row_that_took_its_place() {
        let (mut app, t0) = started();
        app.update(Msg::Key(KeyPress::SelectDown));
        assert_eq!(app.selected(), Some(2), "api, at index 1");

        // api dies; web and worker remain. Index 1 is now worker.
        app.update(Msg::Snapshot {
            rows: vec![
                sheep(1, "web", ProcStatus::Online),
                sheep(3, "worker", ProcStatus::Online),
            ],
            at: t0,
        });
        assert_eq!(app.selected(), Some(3), "the row that took index 1");

        // The LAST row dying clamps rather than leaving the cursor past the end.
        app.update(Msg::Key(KeyPress::SelectLast));
        assert_eq!(app.selected(), Some(3));
        app.update(Msg::Snapshot {
            rows: vec![sheep(1, "web", ProcStatus::Online)],
            at: t0,
        });
        assert_eq!(app.selected(), Some(1));

        // An empty flock selects nothing at all, rather than an id that is gone.
        app.update(Msg::Snapshot {
            rows: vec![],
            at: t0,
        });
        assert_eq!(app.selected(), None);
        assert_eq!(app.selected_index(), None);
    }

    /// fails if moving the selection stops asking for a feed refresh, or
    /// starts asking for one when the selection did not move. The second half
    /// is the one that matters: `RefreshSelected` reads two files off disk
    /// and asks the shepherd for lambs, and a held `k` at the top of the
    /// flock must not do that once per keypress.
    #[test]
    fn a_selection_that_moves_refreshes_the_feed_and_one_that_cannot_does_not() {
        let (mut app, _) = started();
        assert_eq!(
            app.update(Msg::Key(KeyPress::SelectDown)),
            Effect::RefreshSelected
        );
        assert_eq!(
            app.update(Msg::Key(KeyPress::SelectFirst)),
            Effect::RefreshSelected
        );
        assert_eq!(
            app.update(Msg::Key(KeyPress::SelectUp)),
            Effect::None,
            "already at the top: nothing moved, so nothing is re-read"
        );
        assert_eq!(
            app.update(Msg::Key(KeyPress::SelectLast)),
            Effect::RefreshSelected
        );
        assert_eq!(
            app.update(Msg::Key(KeyPress::SelectDown)),
            Effect::None,
            "already at the bottom"
        );
    }

    /// fails if a moved selection stops asking for lambs. The pane describes
    /// the selected sheep, so the trigger is the selection changing and
    /// nothing else.
    #[test]
    fn moving_the_selection_asks_for_lambs() {
        let (mut app, _t0) = started();
        assert_eq!(
            app.update(Msg::Key(KeyPress::SelectDown)),
            Effect::RefreshSelected
        );
    }

    /// fails if the two-second listing starts asking for lambs, which is the
    /// whole cost decision inverted. `with_lambs`'s own doc says `ListFlock`
    /// declines the walk because a flock listing is what an operator leaves
    /// running in a loop; a dashboard putting a full machine enumeration on a
    /// fixed 2s clock, times however many lookout windows are open, is the
    /// daemon paying exactly the cost its own code was written to avoid.
    #[test]
    fn a_snapshot_refreshes_the_feed_and_does_not_ask_for_lambs() {
        let (mut app, t0) = started();
        assert_eq!(
            app.update(Msg::Snapshot {
                rows: vec![sheep(1, "web", ProcStatus::Online)],
                at: t0,
            }),
            Effect::RefreshFeed
        );
    }

    /// fails if a frozen dashboard asks the shepherd for anything. Inherited
    /// from `select_at`'s shipped rule rather than restated: the link is gone,
    /// so there is nothing to ask.
    #[test]
    fn nothing_is_requested_while_the_link_is_lost() {
        let (mut app, _t0) = started();
        app.update(Msg::Frozen {
            at_local: "2026-08-16 09:00:00".to_string(),
        });
        assert_eq!(app.update(Msg::Key(KeyPress::SelectDown)), Effect::None);
    }

    /// fails if a snapshot stops refreshing the feed. This is the whole
    /// cadence: the feed has no timer of its own, and inherits the two-second
    /// listing, the drop repair, the lag repair and `r` from this one line.
    /// It must NOT fire once the link is lost — a frozen dashboard re-reading
    /// log files would be the one thing on screen still moving.
    #[test]
    fn a_snapshot_refreshes_the_feed_unless_the_link_is_frozen() {
        let (mut app, t0) = started();
        assert_eq!(
            app.update(Msg::Snapshot {
                rows: vec![sheep(1, "web", ProcStatus::Online)],
                at: t0
            }),
            Effect::RefreshFeed
        );
        app.update(Msg::Frozen {
            at_local: "2026-08-14 14:32:07".to_string(),
        });
        assert_eq!(
            app.update(Msg::Snapshot {
                rows: vec![sheep(1, "web", ProcStatus::Online)],
                at: t0
            }),
            Effect::None,
            "a frozen dashboard does not re-read anything"
        );
    }

    /// fails if a frozen dashboard reads a log file. The snapshot path
    /// freezes for free — no snapshots arrive after `Msg::Frozen` — but the
    /// KEY path does not, and `j` on a frozen screen would repaint the feed
    /// with content newer than the banner saying the values are frozen as of
    /// 14:32:07. That is the contradiction on one frame that design decision 7
    /// refuses for the host strip, and 12a's
    /// `the_frozen_frame_does_not_move_however_long_the_link_stays_gone`
    /// cannot catch it, because it presses no keys.
    ///
    /// The cursor still MOVES: the detail pane re-rendering from the frozen
    /// listing is the operator reading data already on the frame, which is
    /// allowed. It is touching the disk that is not.
    #[test]
    fn a_frozen_dashboard_moves_the_cursor_without_touching_a_file() {
        let (mut app, _) = started();
        app.update(Msg::Frozen {
            at_local: "2026-08-14 14:32:07".to_string(),
        });

        assert_eq!(
            app.update(Msg::Key(KeyPress::SelectDown)),
            Effect::None,
            "no file is read once the link is lost"
        );
        assert_eq!(app.selected(), Some(2), "but the cursor moved anyway");
        assert_eq!(app.update(Msg::Key(KeyPress::SelectLast)), Effect::None);
        assert_eq!(app.selected(), Some(3));
    }

    /// fails if a dropped or lagged frame stops triggering an immediate poll.
    /// The drop carries no information about what was lost; the only repair
    /// is to ask the shepherd what things look like now — bark's own reason,
    /// and the reason this dashboard does not go silently wrong under load.
    #[test]
    fn a_drop_and_a_lag_both_ask_for_an_immediate_poll() {
        let (mut app, _) = started();
        assert_eq!(
            app.update(Msg::Event(BusEvent::Dropped { count: 12 })),
            Effect::PollNow
        );
        assert_eq!(app.update(Msg::BusLagged { count: 3 }), Effect::PollNow);
        assert_eq!(
            app.update(Msg::Event(BusEvent::Process {
                event: ProcessEventKind::Online,
                info: sheep(1, "web", ProcStatus::Online),
                manually: false,
                at_ms: 0,
            })),
            Effect::None,
            "an ordinary event needs no repair"
        );
    }

    /// fails if the two drop conditions stop being told apart. `Dropped` is the
    /// SHEPHERD's outbound queue overflowing; `Lagged` is this process failing
    /// to read its own socket fast enough. They live on opposite ends of the
    /// connection, and an operator cannot tell which end to investigate if the
    /// notice reads the same. `bleats` pins the same distinction.
    #[test]
    fn a_shepherd_side_drop_and_a_local_lag_read_differently() {
        let (mut app, _) = started();
        app.update(Msg::Event(BusEvent::Dropped { count: 12 }));
        let shepherd_side = app.notice().expect("a drop leaves a notice").to_string();
        app.update(Msg::BusLagged { count: 3 });
        let local = app.notice().expect("a lag leaves a notice").to_string();

        assert!(shepherd_side.contains("the shepherd dropped"));
        assert!(local.contains("lookout fell behind"));
        assert_ne!(shepherd_side, local);
    }

    /// fails if the uptime column stops advancing between polls. A dashboard
    /// whose uptime only moves on the 2s poll reads as frozen when it is not.
    #[test]
    fn a_running_sheeps_uptime_advances_with_the_heartbeat() {
        let (mut app, t0) = started();
        assert_eq!(app.uptime_ms(app.rows()[0].info.id), Some(60_000));
        app.update(Msg::Tick {
            now: t0 + Duration::from_secs(5),
        });
        assert_eq!(app.uptime_ms(1), Some(65_000));
    }

    /// fails if a FROZEN dashboard keeps counting. This is the specific lie
    /// this whole state exists to avoid: the shepherd is gone, so nothing on
    /// screen is known to still be true, and an UPTIME column that keeps
    /// ticking asserts second by second that a process is still running when
    /// nothing can see it.
    #[test]
    fn a_frozen_dashboard_stops_the_uptime_clock() {
        let (mut app, t0) = started();
        app.update(Msg::Tick {
            now: t0 + Duration::from_secs(5),
        });
        app.update(Msg::Frozen {
            at_local: "2026-08-14 14:32:07".to_string(),
        });
        let at_freeze = app.uptime_ms(1);
        app.update(Msg::Tick {
            now: t0 + Duration::from_secs(400),
        });
        assert_eq!(
            app.uptime_ms(1),
            at_freeze,
            "the clock stopped with the link"
        );
        assert_eq!(at_freeze, Some(65_000));
    }

    /// fails if a sheep that is not running has its uptime animated. A stopped
    /// sheep's `uptime_ms` is a historical fact — how long it ran — and
    /// advancing it invents one.
    #[test]
    fn a_stopped_sheeps_uptime_does_not_advance() {
        let t0 = Instant::now();
        let mut app = App::new(
            Palette::detect(None, None, None),
            Control::ReadOnly,
            "/home/rin/.shep".to_string(),
            t0,
        );
        app.update(Msg::Snapshot {
            rows: vec![sheep(1, "web", ProcStatus::Stopped)],
            at: t0,
        });
        app.update(Msg::Tick {
            now: t0 + Duration::from_secs(30),
        });
        assert_eq!(app.uptime_ms(1), Some(60_000));
    }

    /// fails if the gate is checked for one key and not the others. Replaces
    /// `the_stop_key_refuses_in_both_control_states`: that test's whole
    /// subject was that `x` refuses in BOTH gate states, and half of that
    /// stopped being true this task — behind an open gate `x` now arms. Its
    /// read-only half is strictly subsumed here, which covers all three
    /// verbs; its control-enabled half moved out and became
    /// `an_action_key_arms_a_confirm_and_sends_nothing` below.
    #[test]
    fn every_action_key_refuses_while_the_gate_is_closed() {
        for verb in [ActionVerb::Stop, ActionVerb::Restart, ActionVerb::Reload] {
            let (mut app, _t0) = started();
            app.update(Msg::Key(KeyPress::Action(verb)));
            assert!(
                app.action().is_none(),
                "{verb:?} armed behind a closed gate"
            );
            assert_eq!(
                app.notice().map(ToString::to_string).as_deref(),
                Some("read-only: actions need --allow-control"),
                "{verb:?}"
            );
        }
    }

    /// fails if lookout learns to exit on its own. A `DaemonShutdown` is a
    /// notice here, where in `bleats` it precedes a clean exit — the whole
    /// point of Rin's ruling is that a standing dashboard admits it is stale
    /// rather than vanishing. Only `q` quits.
    #[test]
    fn nothing_but_a_keypress_quits() {
        let (mut app, _) = started();
        for msg in [
            Msg::Event(BusEvent::DaemonShutdown),
            Msg::Event(BusEvent::Dropped { count: 1 }),
            Msg::BusLagged { count: 1 },
            Msg::Retrying { attempt: 5 },
            Msg::Frozen {
                at_local: "2026-08-14 14:32:07".to_string(),
            },
        ] {
            assert_ne!(app.update(msg), Effect::Quit);
        }
        assert_eq!(app.update(Msg::Key(KeyPress::Quit)), Effect::Quit);
    }

    /// fails if `Esc` starts quitting out from under a filter. It is the one
    /// key whose meaning depends on state, which is acceptable only because
    /// the status bar reads `esc clear` for as long as that is in force and
    /// the title keeps `2 of 4 in the flock` on screen.
    ///
    /// **Deviation from the plan's literal (Task 2, Step 2.2):** the plan
    /// asserts `Effect::RefreshFeed` here. Traced against `reseat`/
    /// `set_filter` (both shipped, unchanged, in Task 1): clearing a filter
    /// only WIDENS the visible set, so a selection valid under the narrower
    /// filter is — by construction — still valid under the wider one.
    /// `reseat`'s very first line (`if self.selected_index().is_some() {
    /// return false; }`) therefore always takes the no-reseat path on any
    /// widening change, for every fixture, not just this one — there is no
    /// selected sheep for which clearing a filter could report a move. Empty
    /// filter's own doc says `RefreshFeed` rides "a selection that actually
    /// moved"; here it correctly did not, so `Effect::None` is the honest
    /// answer, not a bug this task should paper over by nudging `reseat`.
    #[test]
    fn esc_clears_the_filter_instead_of_quitting_while_one_is_set() {
        let mut app = filtered("web");
        assert_eq!(app.update(Msg::Key(KeyPress::Escape)), Effect::None);
        assert_eq!(app.filter(), "");
        assert_eq!(app.rows().len(), 4);
    }

    /// fails if the clear becomes unconditional, which is the mirror bug: `q`
    /// and Ctrl-C quit from every non-editing state, and so does `Esc` when
    /// there is no filter to take away first.
    #[test]
    fn esc_still_quits_with_no_filter_set() {
        let (mut app, _t0) = started();
        assert_eq!(app.update(Msg::Key(KeyPress::Escape)), Effect::Quit);
    }

    /// fails if typing into the box stops narrowing the table live, or if
    /// `Enter` narrows it a second time. The design's `filter_editing` frame
    /// is mid-type with the table already narrowed; applying only changes
    /// which keys mean what.
    #[test]
    fn the_table_narrows_while_the_query_is_still_being_typed() {
        let (mut app, _t0) = started();
        app.update(Msg::Key(KeyPress::FilterStart));
        assert_eq!(app.mode(), InputMode::Text);
        for letter in ['w', 'e', 'b'] {
            app.update(Msg::Key(KeyPress::FilterChar(letter)));
        }
        assert_eq!(app.rows().len(), 1, "narrowed before Enter");
        app.update(Msg::Key(KeyPress::FilterApply));
        assert_eq!(app.mode(), InputMode::Normal);
        assert_eq!(
            app.rows().len(),
            1,
            "and applying changed nothing but the mode"
        );
    }

    /// fails if backspace stops widening the table again.
    #[test]
    fn backspace_widens_the_table_back_out() {
        let (mut app, _t0) = started();
        app.update(Msg::Key(KeyPress::FilterStart));
        app.update(Msg::Key(KeyPress::FilterChar('w')));
        app.update(Msg::Key(KeyPress::FilterChar('z')));
        assert_eq!(app.rows().len(), 0);
        app.update(Msg::Key(KeyPress::FilterBackspace));
        assert_eq!(
            app.rows().len(),
            2,
            "wz became w, which matches web and worker"
        );
    }

    /// fails if abandoning the box leaves the filter behind. `Esc` while
    /// editing clears and leaves; the two halves are one action.
    #[test]
    fn esc_while_editing_clears_the_filter_and_leaves_the_box() {
        let (mut app, _t0) = started();
        app.update(Msg::Key(KeyPress::FilterStart));
        app.update(Msg::Key(KeyPress::FilterChar('w')));
        app.update(Msg::Key(KeyPress::FilterAbandon));
        assert_eq!(app.mode(), InputMode::Normal);
        assert_eq!(app.filter(), "");
        assert_eq!(app.rows().len(), 3);
    }

    /// fails if a notice survives into the filter box. Two things keep the
    /// query visible and this is one of them: opening the box takes any
    /// standing notice down (the `self.notice = None` at the top of
    /// `on_key`'s normal branch, which `FilterStart` goes through), and slot
    /// 2 of the bar keeps a notice raised LATER, while the box is open, from
    /// covering it. The second is what actually matters, because
    /// `Msg::BusLagged`, `BusEvent::Dropped` and `BusEvent::DaemonShutdown`
    /// all raise notices with no keypress involved and keep arriving while
    /// somebody types; see the phase plan's "Shapes the design named" #2.
    /// This test pins the first, which is what stops a stale notice
    /// reappearing the instant the box closes.
    #[test]
    fn opening_the_filter_takes_a_notice_off_the_bar() {
        let (mut app, _t0) = started();
        app.update(Msg::Event(BusEvent::Dropped { count: 3 }));
        assert!(app.notice().is_some());
        app.update(Msg::Key(KeyPress::FilterStart));
        assert!(app.notice().is_none(), "the box is what the bar shows now");
    }

    /// fails if a notice raised WHILE the box is open is destroyed rather
    /// than deferred. The rejected fix for the same problem was to clear
    /// `self.notice` at the top of `on_text_key`; it would lose a
    /// `DaemonShutdown` because the operator happened to be mid-word. The bar
    /// hides the notice under the box (slot 2) and this pins that the notice
    /// itself is still there to be shown when the box closes.
    #[test]
    fn a_notice_raised_while_typing_is_deferred_and_not_destroyed() {
        let (mut app, _t0) = started();
        app.update(Msg::Key(KeyPress::FilterStart));
        app.update(Msg::Key(KeyPress::FilterChar('w')));
        app.update(Msg::Event(BusEvent::DaemonShutdown));
        app.update(Msg::Key(KeyPress::FilterChar('e')));
        assert!(
            app.notice().is_some(),
            "typing did not wipe the shepherd's announcement"
        );
        assert_eq!(app.filter(), "we", "and the box kept the query");
    }

    /// fails if a reconnect leaves the dashboard in a state that says nothing.
    /// `Retrying` has to be visible while it is happening — an operator
    /// watching a shepherd restart should see it, and should see it clear.
    #[test]
    fn the_link_state_walks_live_to_retrying_to_lost_and_back() {
        let (mut app, t0) = started();
        assert_eq!(app.link(), &Link::Live);

        app.update(Msg::Retrying { attempt: 1 });
        assert_eq!(app.link(), &Link::Retrying { attempt: 1 });

        app.update(Msg::Relinked);
        assert_eq!(app.link(), &Link::Live);

        app.update(Msg::Retrying { attempt: 5 });
        app.update(Msg::Frozen {
            at_local: "2026-08-14 14:32:07".to_string(),
        });
        assert_eq!(
            app.link(),
            &Link::Lost {
                at_local: "2026-08-14 14:32:07".to_string()
            }
        );

        // A late snapshot after the freeze must not silently unfreeze it: the
        // link task has ended, so there is nothing left to produce one, and
        // accepting it would be accepting a message that cannot exist.
        app.update(Msg::Snapshot {
            rows: vec![],
            at: t0,
        });
        assert!(matches!(app.link(), Link::Lost { .. }));
    }

    /// fails if selecting up at the first row or down at the last one wraps or
    /// panics. Clamping is the choice: wrapping a two-hundred-sheep flock from
    /// the last row to the first on one keypress loses the operator's place
    /// with nothing to undo it.
    #[test]
    fn the_selection_clamps_at_both_ends() {
        let (mut app, _) = started();
        for _ in 0..10 {
            app.update(Msg::Key(KeyPress::SelectUp));
        }
        assert_eq!(
            app.selected_index(),
            Some(0),
            "up past the first row stays on it"
        );
        for _ in 0..10 {
            app.update(Msg::Key(KeyPress::SelectDown));
        }
        assert_eq!(
            app.selected_index(),
            Some(2),
            "down past the last row stays on it"
        );
        app.update(Msg::Key(KeyPress::SelectFirst));
        assert_eq!(app.selected_index(), Some(0));
        app.update(Msg::Key(KeyPress::SelectLast));
        assert_eq!(app.selected_index(), Some(2));
    }

    /// fails if `r` stops asking for a poll while the link is live, or starts
    /// pretending to ask once it is frozen. It is the one key that does I/O,
    /// and it is what an operator presses when they do not believe the screen
    /// — so it has to be honest in both directions. The link task ENDS at the
    /// freeze (design decision 8), which drops its poll receiver, so an
    /// `Effect::PollNow` after that is a `try_send` into a closed channel:
    /// the operator presses the key and gets silence with no reason given.
    #[test]
    fn refresh_polls_while_live_and_says_why_it_cannot_once_frozen() {
        let (mut app, _) = started();
        assert_eq!(app.update(Msg::Key(KeyPress::Refresh)), Effect::PollNow);

        app.update(Msg::Frozen {
            at_local: "2026-08-14 14:32:07".to_string(),
        });
        assert_eq!(
            app.update(Msg::Key(KeyPress::Refresh)),
            Effect::None,
            "there is no link task left to ask"
        );
        let notice = app.notice().expect("a refusal is a notice").to_string();
        assert!(notice.contains("the shepherd is gone"));
        assert!(notice.contains("nothing left to ask"));
    }

    /// fails if a notice outlives the keypress after it. A stale refusal still
    /// on screen a minute later is read as a live one.
    #[test]
    fn the_next_keypress_clears_the_notice() {
        let (mut app, _) = started();
        app.update(Msg::Key(KeyPress::Action(ActionVerb::Stop)));
        assert!(app.notice().is_some());
        app.update(Msg::Key(KeyPress::SelectDown));
        assert!(app.notice().is_none());
    }

    /// fails if a frozen dashboard accepts a host reading. The strip reads
    /// THIS machine, which lookout can still see after the shepherd dies — so
    /// this is the one pane that could keep ticking, and one line ticking over
    /// on a screen whose banner says the values are frozen as of 14:32:07 is a
    /// contradiction on the same frame.
    ///
    /// Asserted in the reducer, which is the single place the rule lives:
    /// `run_ui`'s heartbeat samples unconditionally and this arm decides
    /// whether the dashboard is allowed to believe it, exactly as
    /// `Msg::Tick` and the uptime clock already work.
    #[test]
    fn a_frozen_dashboard_ignores_a_host_sample() {
        let (mut app, _) = started();
        app.update(Msg::Host {
            sample: Some(super::super::source::HostSample {
                load: (2.31, 4.10, 3.88),
                cores: Some(10),
                memory_total_bytes: 32 << 30,
                memory_used_bytes: 12 << 30,
                uptime_seconds: 600,
            }),
        });
        assert!(app.host().is_some(), "a live dashboard takes the sample");

        app.update(Msg::Frozen {
            at_local: "2026-08-14 14:32:07".to_string(),
        });
        let frozen = app.host();
        assert_eq!(app.update(Msg::Host { sample: None }), Effect::None);
        assert_eq!(app.host(), frozen, "the last values stay, unchanged");
        assert!(
            !app.host_unsupported(),
            "and a refused sample changes no flag"
        );
    }

    /// fails if `Msg::Bleats` starts asking for another refresh. A reducer
    /// that answered its own feed update with `Effect::RefreshFeed` would
    /// spin the UI task at full tilt, re-reading two files as fast as the
    /// loop can go — the one recursion this design can have.
    #[test]
    fn applying_a_tail_does_not_ask_for_another_one() {
        let (mut app, _) = started();
        assert_eq!(
            app.update(Msg::Bleats {
                tail: super::super::tail::Tail::default()
            }),
            Effect::None
        );
    }

    /// fails if a frozen dashboard applies a coalesced feed read that landed
    /// after the freeze. `refresh_feed` already refuses to ISSUE a read once
    /// the link is `Link::Lost`, but `run_ui`'s coalesced read is armed by a
    /// `feed_dirty` flag set BEFORE the freeze — a read requested a moment
    /// before `Msg::Frozen` can still be in flight when it lands, and without
    /// a guard here the `Msg::Bleats` arm would apply it anyway. That is the
    /// same contradiction-on-one-frame design decision 7 already refuses for
    /// the host strip and the uptime clock; the single enforcement point for
    /// the feed has to live in this arm; nothing upstream can catch a read
    /// that was already in flight when the freeze landed.
    #[test]
    fn a_frozen_dashboard_ignores_a_bleats_tail_in_flight_at_the_freeze() {
        let (mut app, _) = started();
        let live_tail = super::super::tail::Tail {
            lines: vec![super::super::tail::TailLine {
                stream: super::super::tail::Stream::Out,
                text: "read before the freeze".to_string(),
            }],
            ..Default::default()
        };
        app.update(Msg::Bleats {
            tail: live_tail.clone(),
        });
        assert_eq!(app.feed(), &live_tail, "a live dashboard takes the tail");

        app.update(Msg::Frozen {
            at_local: "2026-08-14 14:32:07".to_string(),
        });

        let in_flight_tail = super::super::tail::Tail {
            lines: vec![super::super::tail::TailLine {
                stream: super::super::tail::Stream::Out,
                text: "read after the freeze".to_string(),
            }],
            ..Default::default()
        };
        assert_eq!(
            app.update(Msg::Bleats {
                tail: in_flight_tail
            }),
            Effect::None
        );
        assert_eq!(
            app.feed(),
            &live_tail,
            "the tail read after the freeze must not reach the rendered frame"
        );
    }

    /// A dashboard whose filter is set without any keymap involved. Task 2
    /// wires `/` to this; this task proves the sequence underneath it.
    ///
    /// Four sheep, two of which contain `web`: `web` at id 1 and `web-worker`
    /// at id 3, with `api` at id 2 sitting BETWEEN them in the map. The gap is
    /// deliberate. It is what makes `j` stepping over a hidden row a
    /// falsifiable claim rather than one a contiguous fixture would pass by
    /// accident.
    fn filtered(query: &str) -> App {
        let t0 = Instant::now();
        let mut app = App::new(
            Palette::detect(None, None, None),
            Control::ReadOnly,
            "/home/rin/.shep".to_string(),
            t0,
        );
        app.update(Msg::Snapshot {
            rows: vec![
                sheep(1, "web", ProcStatus::Online),
                sheep(2, "api", ProcStatus::Online),
                sheep(3, "web-worker", ProcStatus::Online),
                sheep(4, "cron", ProcStatus::Online),
            ],
            at: t0,
        });
        app.set_filter(query.to_string());
        app
    }

    /// fails if the table stops narrowing, or if the flock's real size stops
    /// being available beside the narrowed one. The title reads both numbers
    /// and a title that could only read the narrowed one would understate the
    /// flock, which is the same confident wrong number the `-` CPU cell and
    /// the frozen uptime rule exist to prevent.
    #[test]
    fn a_filter_narrows_the_rows_and_leaves_the_real_size_readable() {
        let app = filtered("web");
        assert_eq!(app.rows().len(), 2, "web and web-worker");
        assert_eq!(app.flock_len(), 4, "the flock did not get smaller");
    }

    /// fails if the filter matches whole names instead of substrings, which is
    /// precisely the failure the CLI's selector grammar would have had:
    /// `ProcessSelector`'s `Name` compares with `==`, so typing `w`, `we`,
    /// `web` toward `web-worker` matches nothing at every step.
    #[test]
    fn the_filter_matches_a_substring_and_not_a_whole_name() {
        assert_eq!(filtered("wor").rows().len(), 1, "web-worker, by its middle");
        assert_eq!(filtered("w").rows().len(), 2);
    }

    /// fails if either `to_lowercase` is dropped. Both directions, because
    /// dropping one of the two leaves the other test passing.
    #[test]
    fn the_filter_ignores_case_in_both_directions() {
        let t0 = Instant::now();
        let mut app = App::new(
            Palette::detect(None, None, None),
            Control::ReadOnly,
            "/home/rin/.shep".to_string(),
            t0,
        );
        app.update(Msg::Snapshot {
            rows: vec![sheep(1, "WebEdge", ProcStatus::Online)],
            at: t0,
        });
        app.set_filter("webedge".to_string());
        assert_eq!(
            app.rows().len(),
            1,
            "a lowercase query against a mixed name"
        );
        app.set_filter("WEBEDGE".to_string());
        assert_eq!(app.rows().len(), 1, "and an uppercase one");
    }

    /// fails if `select_by` walks the whole flock again. This is the whole
    /// point of the task: `j` from the first visible row must land on the
    /// second VISIBLE row, not on whatever id happens to sit next in the map.
    #[test]
    fn j_and_k_step_only_over_visible_rows() {
        let mut app = filtered("web");
        assert_eq!(app.selected(), Some(1), "the first visible sheep");
        app.update(Msg::Key(KeyPress::SelectDown));
        assert_eq!(app.selected(), Some(3), "web-worker, skipping api at id 2");
        app.update(Msg::Key(KeyPress::SelectDown));
        assert_eq!(app.selected(), Some(3), "clamped at the last visible row");
        app.update(Msg::Key(KeyPress::SelectUp));
        assert_eq!(app.selected(), Some(1));
    }

    /// fails if `SelectLast` measures the flock rather than the visible set.
    #[test]
    fn select_last_lands_on_the_last_visible_row() {
        let mut app = filtered("web");
        app.update(Msg::Key(KeyPress::SelectLast));
        assert_eq!(app.selected(), Some(3), "web-worker, not cron at id 4");
    }

    /// fails if a filter that hides the selection snaps to row 0, or drops the
    /// selection entirely while rows are still visible. `reseat`'s shipped
    /// rule is that a lost selection falls to whatever now occupies the same
    /// POSITION, clamped: snapping to the top would throw an operator to the
    /// start of a two hundred sheep flock for typing one more character.
    #[test]
    fn a_filter_that_hides_the_selection_clamps_to_the_nearest_visible_row() {
        let mut app = filtered("");
        app.update(Msg::Key(KeyPress::SelectLast));
        assert_eq!(app.selected(), Some(4), "cron, position 3 of 4");
        app.set_filter("web".to_string());
        assert_eq!(
            app.selected(),
            Some(3),
            "position 3 clamps to the last visible row, which is web-worker"
        );
    }

    /// fails if nothing-matches leaves the selection pointing at a hidden
    /// sheep. Every pane below the table describes the selection, so a
    /// selection nobody can see is four panes describing a sheep that is not
    /// on screen.
    #[test]
    fn nothing_visible_means_nothing_selected() {
        let app = filtered("zzz");
        assert_eq!(app.rows().len(), 0);
        assert_eq!(app.selected(), None);
        assert!(app.selected_row().is_none());
        assert_eq!(app.flock_len(), 4, "the flock is still four sheep");
    }

    /// fails if a snapshot clears the filter, or rebuilds the table from the
    /// unfiltered map. The two-second `ListFlock` reply REPLACES `self.flock`
    /// wholesale and is by far the most frequent message this reducer sees, so
    /// a regression here would make the filter appear to work for two seconds
    /// and then silently widen the table under an operator who is still
    /// reading it, with the title's `2 of 4` the only thing left saying a
    /// filter is on.
    #[test]
    fn a_filter_survives_the_two_second_snapshot() {
        let mut app = filtered("web");
        let t1 = Instant::now();
        app.update(Msg::Snapshot {
            rows: vec![
                sheep(1, "web", ProcStatus::Online),
                sheep(2, "api", ProcStatus::Online),
                sheep(3, "web-worker", ProcStatus::Online),
                sheep(4, "cron", ProcStatus::Online),
            ],
            at: t1,
        });
        assert_eq!(app.filter(), "web", "the snapshot did not clear it");
        assert_eq!(app.rows().len(), 2, "and did not widen the table");
        assert_eq!(app.flock_len(), 4);
    }

    /// fails if clearing the filter does not bring the whole flock back, or
    /// leaves the selection unseated. An empty query is the same as no filter,
    /// which is also what `Enter` on an empty box has to mean.
    #[test]
    fn an_empty_query_is_the_same_as_no_filter() {
        let mut app = filtered("zzz");
        app.set_filter(String::new());
        assert_eq!(app.rows().len(), 4);
        assert_eq!(app.selected(), Some(1), "seated again");
    }

    /// fails if the three states `ProcessInfo::lambs` distinguishes get
    /// collapsed. The wire type keeps them apart on purpose: `None` means this
    /// reply did not walk, `Some(vec![])` means it walked and found nothing,
    /// and the pane says different sentences for each because they are
    /// different facts about the machine.
    #[test]
    fn a_lamb_reply_records_which_of_the_three_states_it_saw() {
        let (mut app, t0) = started();
        let walked = ProcessInfo::builder(1, "web", ProcStatus::Online)
            .lambs(Some(vec![Lamb::new(48_220, "node")]))
            .build();
        app.update(Msg::Replied {
            sent: Sent::Lambs { id: 1 },
            result: Ok(Response::Described(vec![walked])),
        });
        assert!(matches!(app.lambs_for(1), Some((LambWalk::Walked(lambs), _)) if lambs.len() == 1));

        let empty = ProcessInfo::builder(1, "web", ProcStatus::Online)
            .lambs(Some(Vec::new()))
            .build();
        app.update(Msg::Replied {
            sent: Sent::Lambs { id: 1 },
            result: Ok(Response::Described(vec![empty])),
        });
        assert!(matches!(app.lambs_for(1), Some((LambWalk::Walked(lambs), _)) if lambs.is_empty()));

        let unwalked = ProcessInfo::builder(1, "web", ProcStatus::Stopped).build();
        app.update(Msg::Replied {
            sent: Sent::Lambs { id: 1 },
            result: Ok(Response::Described(vec![unwalked])),
        });
        assert!(matches!(app.lambs_for(1), Some((LambWalk::NotWalked, _))));
        let _ = t0;
    }

    /// fails if a reading taken for one sheep is shown against another. The
    /// reading carries the id it was taken for and the pane asks by id, so a
    /// request dropped by a full channel and a reply for the previous
    /// selection both read as "not read yet" with no second field to track
    /// them.
    #[test]
    fn a_reading_for_another_sheep_reads_as_not_read_yet() {
        let (mut app, _t0) = started();
        app.update(Msg::Replied {
            sent: Sent::Lambs { id: 1 },
            result: Ok(Response::Described(vec![
                ProcessInfo::builder(1, "web", ProcStatus::Online)
                    .lambs(Some(vec![Lamb::new(48_220, "node")]))
                    .build(),
            ])),
        });
        assert!(app.lambs_for(1).is_some());
        assert!(app.lambs_for(2).is_none(), "not this sheep's reading");
    }

    /// fails if a failed lamb fetch steals the status bar. It is a decoration
    /// on a pane, not an operator's action, and the pane already says what it
    /// does not know (A17).
    #[test]
    fn a_failed_lamb_fetch_says_so_in_the_pane_and_raises_no_notice() {
        let (mut app, _t0) = started();
        app.update(Msg::Replied {
            sent: Sent::Lambs { id: 1 },
            result: Err(RequestError::Closed),
        });
        assert!(matches!(app.lambs_for(1), Some((LambWalk::Failed, _))));
        assert!(app.notice().is_none(), "no notice for a decoration");
    }

    /// fails if an unrecognised reply is recorded as a successful walk.
    /// `Response` is `#[non_exhaustive]`; a variant this binary predates is
    /// not a lamb list and must not read as an empty one, which would say
    /// "none found" about a machine nobody looked at.
    #[test]
    fn an_unrecognised_lamb_reply_is_a_failure_and_not_an_empty_walk() {
        let (mut app, _t0) = started();
        app.update(Msg::Replied {
            sent: Sent::Lambs { id: 1 },
            result: Ok(Response::Pong),
        });
        assert!(matches!(app.lambs_for(1), Some((LambWalk::Failed, _))));
    }

    /// fails if a reading landing after a freeze reaches the frame. Same guard
    /// and same reason as `Msg::Bleats`: the fetch is armed before the freeze
    /// can land, so a reply can still be in flight when `Msg::Frozen` arrives,
    /// and content newer than a banner saying the values are frozen is the
    /// contradiction-on-one-frame this dashboard refuses everywhere else.
    #[test]
    fn a_lamb_reply_after_a_freeze_is_refused() {
        let (mut app, _t0) = started();
        app.update(Msg::Frozen {
            at_local: "2026-08-16 09:00:00".to_string(),
        });
        app.update(Msg::Replied {
            sent: Sent::Lambs { id: 1 },
            result: Ok(Response::Described(vec![
                ProcessInfo::builder(1, "web", ProcStatus::Online)
                    .lambs(Some(vec![Lamb::new(48_220, "node")]))
                    .build(),
            ])),
        });
        assert!(
            app.lambs_for(1).is_none(),
            "the frozen frame learned nothing"
        );
    }

    /// fails if the reply's rows are thrown away and the table left to wait
    /// for the next poll. The shepherd's own rows are right there.
    #[test]
    fn an_accepted_stop_upserts_the_rows_the_shepherd_returned() {
        let mut app = allowed();
        app.update(Msg::Key(KeyPress::Action(ActionVerb::Stop)));
        app.update(Msg::Key(KeyPress::Confirm));
        app.update(Msg::Replied {
            sent: Sent::Action {
                verb: ActionVerb::Stop,
                id: 2,
                name: "api".to_string(),
            },
            result: Ok(Response::Stopped(vec![sheep(
                2,
                "api",
                ProcStatus::Stopped,
            )])),
        });
        assert_eq!(
            app.rows()
                .iter()
                .find(|row| row.info.id == 2)
                .map(|row| row.info.status),
            Some(ProcStatus::Stopped),
            "the table shows what the shepherd said, without waiting for a poll"
        );
        assert_eq!(
            app.notice().map(ToString::to_string).as_deref(),
            Some("stop api (id 2): the shepherd stopped it")
        );
        assert!(app.action().is_none(), "the in-flight state cleared");
    }

    /// fails if a reload reply claims the swap finished. `Response::Reloading`
    /// is an ACCEPTANCE, its own doc says so, and the swaps arrive afterwards
    /// on the bus as `process.reload` / `process.reloaded` /
    /// `process.reload_abandoned`, which the table already consumes. A
    /// sentence saying "reloaded" would be the one lie this reply makes easy
    /// to tell.
    #[test]
    fn a_reload_reply_does_not_claim_the_swap_finished() {
        let mut app = allowed();
        app.update(Msg::Key(KeyPress::Action(ActionVerb::Reload)));
        app.update(Msg::Key(KeyPress::Confirm));
        app.update(Msg::Replied {
            sent: Sent::Action {
                verb: ActionVerb::Reload,
                id: 2,
                name: "api".to_string(),
            },
            result: Ok(Response::Reloading(vec![sheep(
                2,
                "api",
                ProcStatus::Online,
            )])),
        });
        let said = app.notice().map(ToString::to_string).unwrap_or_default();
        assert_eq!(
            said,
            "reload api (id 2): accepted, the swaps report themselves as they happen"
        );
        assert!(!said.contains("reloaded"), "got {said:?}");
    }

    /// fails if the daemon's own words are replaced with a canned string. The
    /// message is a sentence a human wrote; `RequestError`'s full `Display`
    /// interpolates the code with `{:?}` and would put a Rust identifier on an
    /// operator's screen.
    #[test]
    fn a_daemon_refusal_reaches_the_bar_in_the_daemons_own_words() {
        let mut app = allowed();
        app.update(Msg::Key(KeyPress::Action(ActionVerb::Restart)));
        app.update(Msg::Key(KeyPress::Confirm));
        app.update(Msg::Replied {
            sent: Sent::Action {
                verb: ActionVerb::Restart,
                id: 2,
                name: "api".to_string(),
            },
            result: Err(RequestError::Rpc(RpcError {
                code: RpcErrorCode::NotFound,
                message: "selector matched no registered sheep".to_string(),
            })),
        });
        let said = app.notice().map(ToString::to_string).unwrap_or_default();
        assert_eq!(
            said,
            "restart api (id 2): selector matched no registered sheep"
        );
        assert!(!said.contains("NotFound"), "no Rust identifiers: {said:?}");
        assert!(app.notice().is_some_and(Notice::is_grave));
    }

    /// fails if a connection that died mid-request reports as anything else.
    #[test]
    fn a_connection_that_died_mid_request_says_so_under_the_same_prefix() {
        let mut app = allowed();
        app.update(Msg::Key(KeyPress::Action(ActionVerb::Stop)));
        app.update(Msg::Key(KeyPress::Confirm));
        app.update(Msg::Replied {
            sent: Sent::Action {
                verb: ActionVerb::Stop,
                id: 2,
                name: "api".to_string(),
            },
            result: Err(RequestError::Closed),
        });
        let said = app.notice().map(ToString::to_string).unwrap_or_default();
        assert!(said.starts_with("stop api (id 2): "), "got {said:?}");
        assert!(said.contains(&RequestError::Closed.to_string()));
    }

    /// fails if a reply this binary does not understand reads as success.
    /// `Response` is `#[non_exhaustive]`, and swallowing an unrecognised
    /// variant into `Ok` is what `flock()` does and what this must not: a
    /// stop that silently reported success while the sheep kept running is the
    /// worst outcome this feature has.
    ///
    /// The second half is the sharper case: the RIGHT SHAPE for the wrong
    /// verb. A `Stopped` answering a `Restart` carries rows and would upsert
    /// happily.
    #[test]
    fn an_unrecognised_reply_says_so_rather_than_reading_as_success() {
        for reply in [
            Response::Pong,
            Response::Stopped(vec![sheep(2, "api", ProcStatus::Stopped)]),
        ] {
            let mut app = allowed();
            app.update(Msg::Key(KeyPress::Action(ActionVerb::Restart)));
            app.update(Msg::Key(KeyPress::Confirm));
            app.update(Msg::Replied {
                sent: Sent::Action {
                    verb: ActionVerb::Restart,
                    id: 2,
                    name: "api".to_string(),
                },
                result: Ok(reply),
            });
            assert_eq!(
                app.notice().map(ToString::to_string).as_deref(),
                Some(
                    "restart api (id 2): the shepherd answered something this lookout does not understand"
                )
            );
            assert!(app.notice().is_some_and(Notice::is_grave));
        }
    }
}