shep 0.8.1

//! Query verbs: `flock`, `describe`, `fold`, `ping`. None mutate the flock,
//! and none autostart: `main` hands each one an already-connected [`Client`].
//!
//! `describe` and `fold` share one shape (`Request::Describe` against a
//! [`SelectorSpec`]); `fold` supplies `SelectorSpec::Fold` directly.
//!
//! `ping` does not ask the daemon for its version and pid: the handshake
//! already answered that, in the
//! [`HelloAck`](shep_core::protocol::HelloAck) [`Client::daemon`] holds. It
//! still issues `Request::Ping` as the liveness check.

use std::collections::BTreeMap;
use std::io::{self, Write as _};
use std::time::Duration;

use crossterm::QueueableCommand as _;
use crossterm::cursor::{Hide, MoveTo, Show};
use crossterm::terminal::{Clear, ClearType};
use shep_client::Client;
use shep_core::paths::ShepPaths;
use shep_core::protocol::{HostUsage, ProcessInfo, Request, Response, SelectorSpec};
#[cfg(test)]
use shep_core::secrets::Resolution;
use shep_core::secrets::{self, SecretRef, SecretView};
use shep_core::status::ProcStatus;
use shep_daemon::snapshot::FlockSnapshot;

use crate::cli::{DogsArgs, FoldArgs, Format, SelectorArgs};
use crate::commands::rpc::{client_error, request_and_render, unexpected_response};
use crate::commands::selector::parse_selector_spec;
use crate::dog_index::{self, AvailableDog, DogSourceKind};
use crate::exit::ExitCode;
use crate::fetch;
use crate::flourish;
use crate::host;
use crate::lookout::term;
use crate::output::width;
use crate::output::{
    AvailableDogRows, DescribedSecret, DogRows, RolledSheep, RolledSheepRows, SecretStatus,
    Streams, emit, emit_described, emit_flock, write_outcome,
};
use crate::shutdown::Interrupt;
use crate::style::Presentation;

/// `describe` and `fold`'s shared body: one `Request::Describe` against
/// `selector`, rendered through [`emit_described`] as the sheep table and
/// each sheep's lamb tree beneath it. `command` is the verb name the output
/// envelope reports.
///
/// `include_secrets` alone gates [`gather_secrets`]: `fold` passes `false`
/// and stays byte-identical to before this section existed, because `fold`
/// is a group view across a selector's sheep, not the single-sheep
/// diagnostic this feature was built for.
///
/// Not routed through [`request_and_render`]: `emit_described` renders one
/// `Vec<ProcessInfo>` into two tables, which no single
/// [`crate::output::Render`] impl can express.
async fn describe_selector(
    client: &Client,
    streams: &mut Streams<'_>,
    paths: &ShepPaths,
    command: &str,
    include_secrets: bool,
    selector: SelectorSpec,
) -> ExitCode {
    match client.request(Request::Describe { selector }).await {
        Ok(Response::Described(procs)) => {
            let secrets = if include_secrets {
                let (rows, unreadable) = gather_secrets(paths, &procs);
                if let Some(error) = unreadable {
                    let message = format!(
                        "the secret store at {} could not be read ({error}), so every \
                         reference below reads as missing whether or not the store holds \
                         a value for it",
                        paths.secrets.display()
                    );
                    streams.aside(SECRET_STORE_UNREADABLE_NOTICE, &message);
                }
                rows
            } else {
                Vec::new()
            };
            let result = emit_described(
                &mut *streams.out,
                streams.fmt,
                command,
                procs,
                streams.style,
                &secrets,
            );
            write_outcome(result)
        }
        Ok(_unrecognised) => unexpected_response(streams),
        Err(err) => client_error(streams, &err),
    }
}

/// Renders one sheep's secret references the way `describe`'s table form
/// would: one line per reference, the reference as written, the environment
/// it resolved in, and a verdict. Never a value: `Resolution::Found`'s
/// payload is read only to tell it apart from a miss.
///
/// The verdict word comes from [`SecretStatus::from_resolution`], the same
/// classifier [`gather_secrets`] uses for the JSON form, so the two can
/// never name a different verdict for the same reference.
///
/// Test-only: `emit_described` renders the actual table now, so this exists
/// to exercise [`SecretStatus::from_resolution`]'s three verdicts in
/// isolation, without a fake daemon connection.
#[cfg(test)]
fn render_describe_secrets(entries: &[(&str, &str, Resolution<'_>)]) -> String {
    let mut rendered = String::new();
    for (reference, environment, resolution) in entries {
        let verdict = SecretStatus::from_resolution(resolution).as_table_word();
        rendered.push_str(&format!("  {reference} ({environment}): {verdict}\n"));
    }
    rendered
}

/// `describe`'s secrets section for `procs`, once per distinct sheep name:
/// the JSON-safe rows [`emit_described`] both serializes beside `data` and
/// renders as the table's own "Secrets for `<name>`:" block.
///
/// Reads three local files rather than asking the shepherd: the muster roll
/// for each name's [`shep_core::config::AppConfig`] (which can trail a
/// config change that has not yet reached disk by
/// `shep_daemon::snapshot`'s debounce window), the operator's own secret
/// store, and the provider cache [`secrets::provider_cache_on_disk`]
/// reads. A namespace whose provider pushed with `persist = false` never
/// reaches that cache, so this can call a namespace uncached when the
/// running shepherd already has it in memory; only the shepherd itself can
/// answer that half.
///
/// A name the roll does not know, or one with no `{{secret:...}}` at all,
/// contributes nothing: this reports references that exist, not a claim
/// that every sheep has one.
///
/// The second half of the answer is why the operator's own store is empty,
/// when it is empty because it would not read. Folding that into an empty
/// store on its own would print `missing` beside every bare reference and
/// send the operator to `shep secret set` for values the store may already
/// hold. Rows are still produced, so a sheep needing nothing still
/// describes; the caller has the [`Streams`] to say so on.
fn gather_secrets(
    paths: &ShepPaths,
    procs: &[ProcessInfo],
) -> (Vec<DescribedSecret>, Option<secrets::SecretError>) {
    let (store, unreadable) = match secrets::all(&paths.secrets) {
        Ok(store) => (store, None),
        Err(error) => (BTreeMap::new(), Some(error)),
    };
    let providers = secrets::provider_cache_on_disk(&paths.secrets_cache);

    let mut json = Vec::new();
    for namer in crate::secret_readers::namers(paths, procs) {
        let view = SecretView::new(namer.environment.clone(), store.clone(), providers.clone());
        for reference in &namer.references {
            let Some(parsed) = SecretRef::parse(reference) else {
                continue;
            };
            json.push(DescribedSecret {
                name: namer.name.clone(),
                reference: reference.clone(),
                environment: namer.environment.clone(),
                status: SecretStatus::from_resolution(&view.resolve(&parsed)),
            });
        }
    }
    (json, unreadable)
}

/// [`emit_notice`](crate::output::emit_notice) code for the warning
/// `describe` prints when the operator's secret store will not read. Not a
/// failure: the sheep still describes.
const SECRET_STORE_UNREADABLE_NOTICE: &str = "secret_store_unreadable";

/// Whatever `flock.json` currently holds, or nothing when it is missing or
/// will not parse.
///
/// Shared by [`flock_from_roll`] and [`gather_secrets`]: both read the
/// muster roll as the best local answer to "what does this app's config
/// look like right now", tolerant of a file this daemon has never written
/// or has fallen behind the live registry by a debounce window.
pub(crate) fn read_roll(paths: &ShepPaths) -> Option<FlockSnapshot> {
    std::fs::read(&paths.snapshot)
        .ok()
        .and_then(|bytes| serde_json::from_slice::<FlockSnapshot>(&bytes).ok())
}

/// `shep flock` when no shepherd answers: the muster roll, marked stopped
///
/// The exit code stays [`ExitCode::DaemonUnreachable`] even though the table
/// looks successful: a monitoring script must not read a dead supervisor as
/// a healthy empty flock. A missing or unreadable roll is not an error.
pub fn flock_from_roll(streams: &mut Streams<'_>, paths: &ShepPaths) -> ExitCode {
    let saved = read_roll(paths);

    let mut sheep: Vec<RolledSheep> = saved
        .map(|roll| {
            roll.apps
                .into_iter()
                .map(|entry| RolledSheep {
                    name: entry.app.name.clone(),
                    instances: entry.instances_running,
                    status: "stopped",
                })
                .collect()
        })
        .unwrap_or_default();
    // The roll's stored order is neither meaningful nor stable. Name is the
    // whole key: one entry per app, so there is no tie to break.
    sheep.sort_unstable_by(|a, b| a.name.cmp(&b.name));

    if streams.fmt == Format::Table {
        let _ = writeln!(
            streams.err,
            "no shepherd running. {}",
            if sheep.is_empty() {
                "nothing in the saved roll either.".to_owned()
            } else {
                format!(
                    "{} in the saved roll at {}:",
                    match sheep.len() {
                        1 => "1 sheep".to_owned(),
                        n => format!("{n} sheep"),
                    },
                    paths.snapshot.display()
                )
            }
        );
    }
    let empty = sheep.is_empty();
    // No table for an empty roll: bare headers over nothing read as a glitch.
    // JSON still gets the empty array, because a script wants one shape.
    if !(empty && streams.fmt == Format::Table) {
        let _ = emit(
            &mut *streams.out,
            streams.fmt,
            "flock",
            RolledSheepRows(sheep),
            streams.style,
        );
    }
    if streams.fmt == Format::Table && !empty {
        let _ = writeln!(streams.err, "`shep muster` brings them back.");
    }
    ExitCode::DaemonUnreachable
}

/// Lists the whole flock: the sheep table, then the dogs table beneath it
/// whenever any dog is registered, with a [`sheep_flourish`] above
///
/// The flourish is gated on `Format::Table` and `streams.style.level.sheep()`
/// and nothing else. Not routed through [`request_and_render`], which renders
/// one [`crate::output::Render`] type per verb rather than two tables from one
/// `Vec<ProcessInfo>`.
pub async fn flock(client: &Client, streams: &mut Streams<'_>) -> ExitCode {
    match client.request(Request::ListFlock).await {
        Ok(Response::Flock(procs)) => {
            // Read before `procs` moves into `emit_flock`.
            let art = (streams.fmt == Format::Table && streams.style.level.sheep())
                .then(|| sheep_flourish(&procs))
                .flatten();
            if let Some(art) = &art {
                let _ = write!(streams.out, "{art}");
            }
            let host = host_usage(client).await;
            if streams.fmt == Format::Table
                && let Some(usage) = host
            {
                // Above the table and separated from it, exactly where a
                // followed listing puts the same line. `Presentation::width`
                // is the terminal's, resolved once at the seam.
                let strip = host::strip(usage, streams.style, streams.style.width);
                let _ = writeln!(streams.out, "{strip}\n");
            }
            write_outcome(emit_flock(
                &mut *streams.out,
                streams.fmt,
                "flock",
                procs,
                host,
                streams.style,
            ))
        }
        Ok(_unrecognised) => unexpected_response(streams),
        Err(err) => client_error(streams, &err),
    }
}

/// What the shepherd says the machine is doing, or `None` where it will not
/// say.
///
/// The two answers inside the `Some` are different claims and a reader
/// renders them differently: a reading, and a platform `sysinfo` cannot read
/// at all.
///
/// The outer `None` is a shepherd that refused the request, which today
/// means one built before `Request::HostUsage` existed. It decodes
/// `Request::Unrecognized` and answers `RpcErrorCode::Unsupported` by name,
/// and the listing then prints no strip: exactly what `shep flock` printed
/// before this existed. Not a warning line, which would fire on every
/// listing for as long as the skew lasts, on the verb an operator types
/// most.
async fn host_usage(client: &Client) -> Option<Option<HostUsage>> {
    match client.request(Request::HostUsage).await {
        Ok(Response::HostUsage(usage)) => Some(usage),
        // A shepherd that answered something else, and one that refused.
        Ok(_unrecognised) => None,
        Err(_unsupported) => None,
    }
}

/// The flourish for one flock listing, or `None` when neither the
/// empty-flock nor the all-asleep state applies
///
/// Dogs are excluded from both checks: the flourish sits beside the sheep
/// table and is a claim about the sheep. [`ProcStatus::Stopping`] does not
/// count as asleep, being reload's transient rather than rest.
fn sheep_flourish(listing: &[ProcessInfo]) -> Option<String> {
    let sheep: Vec<&ProcessInfo> = listing.iter().filter(|p| p.dog.is_none()).collect();
    if sheep.is_empty() {
        return Some(flourish::empty_flock());
    }
    sheep
        .iter()
        .all(|p| p.status == ProcStatus::Stopped)
        .then(|| flourish::all_asleep(sheep.len()))
}

/// `shep flock --follow`: the same listing, painted over itself every
/// `interval` until the operator interrupts it or the shepherd goes.
///
/// Each redraw is one `Request::ListFlock` rendered into a buffer and then
/// written over the screen in a single write. Rendering before clearing is
/// what keeps the terminal from sitting blank for the length of the round
/// trip. The main screen, not the alternate one, so the last frame is still
/// there afterwards. No flourish, which redrawn every second is noise.
///
/// An interrupt ends the follow at [`ExitCode::Success`], a shepherd that
/// goes away mid-follow ends it carrying that refusal's own code. The two
/// must not read as the same thing.
///
/// Why each of those beat its alternative: `docs/decisions.md`, "Following
/// the flock".
pub(crate) async fn flock_follow(
    client: &Client,
    streams: &mut Streams<'_>,
    interval: Duration,
) -> ExitCode {
    let mut interrupt = match Interrupt::install() {
        Ok(interrupt) => interrupt,
        Err(err) => {
            let message = format!("listening for an interrupt: {err}");
            return streams.fail(ExitCode::Failure, &message);
        }
    };
    // The hook covers a panic, the guard covers every other way out of this
    // function, and `term::restore` is documented idempotent because a panic
    // fires both. The guard shows the cursor and stops there: hiding it is
    // the only change this verb makes to the terminal, where `lookout` also
    // takes raw mode and the alternate screen.
    term::install_panic_hook();
    let _cursor = term::RestoreGuard::with_action(|| {
        let _ = crossterm::execute!(io::stdout(), Show);
    });
    let _ = streams.out.queue(Hide);

    let mut ticker = tokio::time::interval(interval);
    // A shepherd slower to answer than the interval would otherwise bank
    // every tick it missed and redraw them back to back the moment it
    // answered. `Delay` measures the next interval from the redraw that just
    // finished, so a slow shepherd slows the cadence instead of bursting it.
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    loop {
        tokio::select! {
            _ = ticker.tick() => {}
            _ = interrupt.recv() => return ExitCode::Success,
        }
        // The request is raced too, so a shepherd that stops answering does
        // not hold the terminal until it does.
        let listing = tokio::select! {
            listing = client.request(Request::ListFlock) => listing,
            _ = interrupt.recv() => return ExitCode::Success,
        };
        let procs = match listing {
            Ok(Response::Flock(procs)) => procs,
            Ok(_unrecognised) => return unexpected_response(streams),
            Err(err) => return client_error(streams, &err),
        };
        let host = tokio::select! {
            host = host_usage(client) => host,
            _ = interrupt.recv() => return ExitCode::Success,
        };
        // One reading, used twice. Measured before the frame is built
        // rather than after it, because the strip fits itself to the window
        // and a window resized between redraws has to be read again. Two
        // readings would let a resize land between them, fitting the strip
        // to the old width and then trimming the frame to the new one.
        let size = crossterm::terminal::size().ok();
        // A width of zero is not a window. `fit_rows` answers that case for
        // itself, further down, by declining to trim at all.
        let columns = size
            .filter(|&(columns, _)| columns > 0)
            .map_or(streams.style.width, |(columns, _)| usize::from(columns));
        let frame = follow_frame(
            procs,
            streams.style,
            host.map(|usage| host::strip(usage, streams.style, columns)),
        );
        let frame = match size {
            Some((columns, rows)) => fit_rows(&frame, columns, rows),
            // A terminal that will not say how big it is gets the frame
            // whole. Scrolling beats hiding a sheep.
            None => frame,
        };
        // Not discarded. A follow whose terminal has gone (the emulator
        // closed, an ssh session dropped) writes into a dead descriptor
        // forever otherwise, and keeps asking the shepherd for a listing it
        // cannot paint. `write_outcome` keeps a broken pipe at
        // `ExitCode::Success`, which is the reader leaving rather than a
        // failure.
        let painted = paint(streams.out, &frame);
        if painted.is_err() {
            return write_outcome(painted);
        }
    }
}

/// One redraw's four writes, as one `io::Result`.
///
/// Separate so the loop above reads as "paint, and stop if that failed"
/// rather than four discarded results in a row.
fn paint(out: &mut dyn io::Write, frame: &str) -> io::Result<()> {
    out.queue(MoveTo(0, 0))?;
    out.queue(Clear(ClearType::FromCursorDown))?;
    write!(out, "{frame}")?;
    out.flush()
}

/// One redraw's worth of text: the host line, then the tables [`flock`]
/// would have printed.
///
/// The host line goes above rather than below so it holds still while the
/// tables under it change length, and so it is the last thing [`fit_rows`]
/// gives up.
fn follow_frame(listing: Vec<ProcessInfo>, style: Presentation, strip: Option<String>) -> String {
    let mut frame = Vec::new();
    if let Some(strip) = strip {
        // Writing to a `Vec` cannot fail, here or below.
        let _ = writeln!(frame, "{strip}\n");
    }
    // `None`: a followed frame is always a table, and the host block rides
    // above it as text rather than inside the payload.
    let _ = emit_flock(&mut frame, Format::Table, "flock", listing, None, style);
    String::from_utf8_lossy(&frame).into_owned()
}

/// Trims `frame` to what a terminal `columns` wide and `rows` tall shows,
/// saying how much it dropped.
///
/// Rows, not lines: a line wider than the terminal wraps onto more than one
/// of them, so counting lines would overrun a narrow window and leave every
/// redraw scrolling. One row is held back for the cursor the redraw leaves
/// behind, and the notice's own height for the notice.
///
/// The notice wraps like anything else, which is why its height is measured
/// rather than assumed to be one. Giving a kept line back to make room was
/// the earlier answer and it does not hold: at 30 columns the notice takes
/// two rows and the line handed back was worth one, so the frame overran by
/// one row and the window scrolled on every redraw. Measured at 12, 16, 20,
/// 24 and 30 columns. Reserving the worst case, every line dropped, costs at
/// most one row more than the final count needs, and no circularity.
///
/// A size of nothing is not a window of nothing. A pty that has never been
/// told how big it is reports zero, and `script(1)` hands `--follow` exactly
/// that: measured 2026-09-12, where trimming to it printed the notice alone,
/// every second, and no flock at all. Sizes that leave no room to trim get
/// the frame whole, the same answer a terminal that will not measure gets.
fn fit_rows(frame: &str, columns: u16, rows: u16) -> String {
    let (columns, budget) = (usize::from(columns), usize::from(rows).saturating_sub(1));
    if columns == 0 || budget == 0 {
        return frame.to_owned();
    }
    let lines: Vec<&str> = frame.lines().collect();
    // Widest the notice can get, since more dropped lines means more digits.
    let reserve = line_rows(&notice(lines.len()), columns);
    let fill = budget.saturating_sub(reserve);
    let mut used = 0;
    let mut kept = 0;
    for line in &lines {
        let height = line_rows(line, columns);
        if used + height > fill {
            break;
        }
        used += height;
        kept += 1;
    }
    if kept == lines.len() {
        return frame.to_owned();
    }
    let mut fitted = lines[..kept].join("\n");
    if kept > 0 {
        fitted.push('\n');
    }
    // `budget - used` rather than the whole notice: a window too narrow to
    // hold it is the one case the reservation above cannot satisfy, and
    // printing it whole overruns the budget and scrolls the screen, which is
    // the single thing this function exists to prevent. Clipped, the frame
    // stays inside its rows and the operator still reads the leading digits,
    // which is the part that says how much is missing.
    fitted.push_str(&clip_rows(
        &notice(lines.len() - kept),
        columns,
        budget.saturating_sub(used),
    ));
    fitted.push('\n');
    fitted
}

/// `text` cut to at most `rows` rows at `columns` wide.
///
/// Only [`fit_rows`]'s notice reaches this, and only on a window too narrow
/// to print it whole.
fn clip_rows(text: &str, columns: usize, rows: usize) -> String {
    let ceiling = rows.saturating_mul(columns);
    if width::visible_width(text) <= ceiling {
        return text.to_owned();
    }
    text.chars().take(ceiling).collect()
}

/// What a trimmed frame says in place of the lines it dropped.
///
/// A function rather than a literal because [`fit_rows`] measures this twice:
/// once at its widest to reserve the rows, and once with the count it settled
/// on.
fn notice(dropped: usize) -> String {
    // Singular is reachable: `dropped` is at least one wherever the notice
    // prints at all, and a follow redraws this once a second.
    let lines = if dropped == 1 { "line" } else { "lines" };
    format!("{dropped} more {lines} than this terminal shows")
}

/// How many terminal rows `line` occupies once it wraps at `columns`.
///
/// An empty line still occupies one.
fn line_rows(line: &str, columns: usize) -> usize {
    width::visible_width(line).div_ceil(columns).max(1)
}

/// Lists the dogs and nothing else: the same `Request::ListFlock` [`flock`]
/// sends, filtered to the entries carrying a `dog` marker
///
/// `args.filter` is a case-insensitive substring match against the dog's
/// name, the one field a running dog and a community-index entry share.
/// Not [`emit_flock`], which would print the sheep table's header row over
/// a dogs-only listing.
pub async fn dogs(client: &Client, streams: &mut Streams<'_>, args: &DogsArgs) -> ExitCode {
    let filter = args.filter.as_deref();
    request_and_render(
        client,
        streams,
        "dogs",
        Request::ListFlock,
        None,
        |response| match response {
            Response::Flock(procs) => Some(DogRows(
                procs
                    .into_iter()
                    .filter(|p| p.dog.is_some())
                    .filter(|p| filter.is_none_or(|f| matches_filter(f, &[&p.name])))
                    .collect(),
            )),
            _ => None,
        },
    )
    .await
}

/// Whether `filter` matches any of `haystacks`, case-insensitively. Shared
/// by [`dogs`] (name alone) and [`available_dogs`] (name, package and
/// description).
fn matches_filter(filter: &str, haystacks: &[&str]) -> bool {
    let filter = filter.to_lowercase();
    haystacks.iter().any(|h| h.to_lowercase().contains(&filter))
}

/// Lists the dogs published in the community index: `shep dogs --available`
///
/// Reaches no [`Client`], so it answers with no shepherd running. Under
/// `Format::Table` a filter matching one dog prints its detail view, and a
/// filter matching nothing prints `no dog matches "<filter>"` and still
/// exits [`ExitCode::Success`]. `--format json` always renders the array.
///
/// # Errors reaching the operator
/// A failure to read or parse the index names the URL and exits
/// [`ExitCode::Failure`]: [`dog_index::IndexError`] carries it on one
/// variant only.
///
/// The one URL not named is one holding an `@`. A dog index URL is a
/// public location, which is why this quotes it at all, but
/// `SHEP_DOG_INDEX` is an operator's own string and nothing stops a
/// password reaching it. This message is built here rather than by
/// [`dog_index::IndexError`], so the refusals inside [`crate::fetch`] do
/// not cover it, and it asks
/// [`fetch::url_for_message`](crate::fetch::url_for_message) rather than
/// deciding for itself: an earlier version asked
/// `url_carries_credentials` instead and printed urls that `parse_url`
/// had just withheld.
pub async fn available_dogs(streams: &mut Streams<'_>, args: &DogsArgs) -> ExitCode {
    let url = dog_index::index_url();
    let index = match dog_index::fetch_index(&url).await {
        Ok(index) => index,
        Err(err) => {
            let message = format!(
                "reading the dog index from {}: {err}",
                fetch::url_for_message(&url)
            );
            return streams.fail(ExitCode::Failure, &message);
        }
    };
    let (skipped, sanitised) = (index.skipped, index.sanitised);

    let filter = args.filter.as_deref();
    let matched: Vec<AvailableDog> = index
        .dogs
        .into_iter()
        .filter(|dog| {
            filter.is_none_or(|f| matches_filter(f, &[&dog.name, &dog.package, &dog.description]))
        })
        .collect();

    let code = if streams.fmt == Format::Table
        && matched.is_empty()
        && let Some(filter) = filter
    {
        let _ = writeln!(streams.out, "no dog matches {filter:?}");
        ExitCode::Success
    } else if streams.fmt == Format::Table
        && let [only] = matched.as_slice()
    {
        write_outcome(render_detail(&mut *streams.out, only))
    } else {
        write_outcome(emit(
            &mut *streams.out,
            streams.fmt,
            "dogs",
            AvailableDogRows(matched),
            streams.style,
        ))
    };

    note_index_costs(streams, skipped, sanitised);
    code
}

/// The clause both notices below end in: the counts describe the fetched
/// document, not the filtered listing they are printed beside.
const INDEX_WIDE: &str = ", across the whole index rather than this listing";

/// Prints [`dog_index::Index::skipped`]/[`dog_index::Index::sanitised`] as
/// footer notices when either is non-zero, whatever the filter matched.
fn note_index_costs(streams: &mut Streams<'_>, skipped: usize, sanitised: usize) {
    if skipped > 0 {
        streams.aside(
            "dogs_skipped",
            &format!(
                "{skipped} entr{} skipped{INDEX_WIDE}",
                if skipped == 1 { "y" } else { "ies" }
            ),
        );
    }
    if sanitised > 0 {
        streams.aside(
            "dogs_sanitised",
            &format!(
                "{sanitised} entr{} contained control characters{INDEX_WIDE}",
                if sanitised == 1 { "y" } else { "ies" }
            ),
        );
    }
}

/// The lone-match affordance [`available_dogs`] prints for `Format::Table`:
/// full detail on one dog, ending in the two copy-pasteable commands an
/// operator needs to adopt it. Never reached from `--format json`.
///
/// # Errors
/// The underlying write failed.
fn render_detail(out: &mut dyn std::io::Write, dog: &AvailableDog) -> std::io::Result<()> {
    writeln!(out, "{} . {} . {}", dog.name, dog.package, dog.category)?;
    writeln!(out, "{}", dog.description)?;
    writeln!(out, "{} . {}", dog.license, dog.repo)?;
    writeln!(out)?;
    writeln!(out, "{}", install_line(&dog.source, &dog.package))?;
    writeln!(
        out,
        "{}",
        adopt_line(&dog.source, &dog.adopt_as, &dog.package)
    )
}

/// The `$ ...` line [`render_detail`] prints for how to build `source`'s
/// binary. [`DogSourceKind::Manual`] carries prose instead of a command, so
/// it prints with no `$`, just the two-space indent the command lines share.
fn install_line(source: &DogSourceKind, package: &str) -> String {
    match source {
        DogSourceKind::Cargo {
            version: Some(version),
        } => {
            format!("  $ cargo install {package} --version {version}")
        }
        DogSourceKind::Cargo { version: None } => format!("  $ cargo install {package}"),
        DogSourceKind::CargoGit { url } => format!("  $ cargo install --git {url}"),
        DogSourceKind::GoInstall { module } => format!("  $ go install {module}@latest"),
        DogSourceKind::Manual { instructions } => format!("  {instructions}"),
    }
}

/// The `$ shep adopt ...` line [`render_detail`] prints
///
/// Built from `adopt_as`, never `name` or `package`: a wrong name ships a
/// command that silently discards the dog's whole config section.
/// [`DogSourceKind::Manual`] has no predictable install path, so its line
/// names the placeholder literally. `--name` is always spelled, since
/// nothing enforces the naming convention on a user-contributed `package`.
fn adopt_line(source: &DogSourceKind, adopt_as: &str, package: &str) -> String {
    match source {
        DogSourceKind::Cargo { .. } | DogSourceKind::CargoGit { .. } => {
            format!("  $ shep adopt ~/.cargo/bin/{package} --name {adopt_as}")
        }
        DogSourceKind::GoInstall { .. } => {
            format!("  $ shep adopt $(go env GOPATH)/bin/{package} --name {adopt_as}")
        }
        DogSourceKind::Manual { .. } => {
            format!("  $ shep adopt <path to the binary> --name {adopt_as}")
        }
    }
}

/// Describes the sheep matching `args.selector` in detail.
pub async fn describe(
    client: &Client,
    streams: &mut Streams<'_>,
    paths: &ShepPaths,
    args: &SelectorArgs,
) -> ExitCode {
    // One pass per target, each its own detail view: `describe` answers with
    // a tree per sheep, so merging them would lose that shape.
    let mut failure: Option<ExitCode> = None;
    for raw in &args.selectors {
        let selector = match parse_selector_spec(streams, raw) {
            Ok(selector) => selector,
            Err(code) => return code,
        };
        let code = describe_selector(client, streams, paths, "describe", true, selector).await;
        if code != ExitCode::Success {
            failure = failure.or(Some(code));
        }
    }
    failure.unwrap_or(ExitCode::Success)
}

/// Lists one fold: `Request::Describe` with `SelectorSpec::Fold(args.name)`,
/// delegating to [`describe_selector`].
pub async fn fold(
    client: &Client,
    streams: &mut Streams<'_>,
    paths: &ShepPaths,
    args: &FoldArgs,
) -> ExitCode {
    describe_selector(
        client,
        streams,
        paths,
        "fold",
        false,
        SelectorSpec::Fold(args.name.clone()),
    )
    .await
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use shep_client::testing::{
        fake_client_answering, fake_client_capturing_envelopes, fake_client_on,
        fake_client_replying_err, fake_client_with_ack, sample_ack, sample_info,
    };
    use shep_core::protocol::{DogSource, RpcErrorCode};

    use super::*;

    /// Bounds every `envelopes.recv()` here: a verb that never reaches the
    /// wire must fail by assertion, not by hanging the job.
    const RECV_TIMEOUT: Duration = Duration::from_secs(5);

    /// fails if a frame that fits gets trimmed anyway. Three lines in a
    /// window with rows to spare come back byte-identical, trailing newline
    /// and all.
    /// fails if the notice's own height stops being reserved. It wraps like
    /// any other line, so at 30 columns it is two rows and the old "hand one
    /// kept line back" reservation bought one. Measured overruns before the
    /// fix: 12x3, 12x5, 16x3, 20x3, 24x3 and 30x3.
    ///
    /// Including the window too narrow to hold the notice whole, which is
    /// the case that used to be skipped here: it is clipped now rather than
    /// allowed to overrun.
    #[test]
    fn a_trimmed_frame_never_overruns_the_rows_it_was_given() {
        let frame: String = (0..40)
            .map(|n| format!("sheep-{n:02}  online  1234  0.5%  12.3 MB  0d 0h 1m\n"))
            .collect();
        for columns in [12u16, 16, 20, 24, 30, 38, 40, 60, 80, 120] {
            for rows in [3u16, 4, 5, 8, 12, 24, 40] {
                let out = fit_rows(&frame, columns, rows);
                let used: usize = out
                    .lines()
                    .map(|line| line_rows(line, usize::from(columns)))
                    .sum();
                let budget = usize::from(rows).saturating_sub(1);
                assert!(
                    used <= budget,
                    "{columns}x{rows}: used {used} rows against a budget of {budget}"
                );
            }
        }
    }

    /// fails if the notice goes back to one spelling. Dropping exactly one
    /// line is the common case on a window one row short, and it redraws
    /// every second.
    #[test]
    fn the_notice_counts_one_dropped_line_in_the_singular() {
        let frame = "one\ntwo\nthree\n";

        // Four rows: one held for the cursor, one for the notice, two for
        // content, so exactly one line drops and the word is singular.
        assert_eq!(
            fit_rows(frame, 80, 4),
            "one\ntwo\n1 more line than this terminal shows\n"
        );
        // Three rows leaves one for content, so two drop and it is plural.
        assert_eq!(
            fit_rows(frame, 80, 3),
            "one\n2 more lines than this terminal shows\n"
        );
    }

    #[test]
    fn a_frame_that_fits_is_left_alone() {
        let frame = "one\ntwo\nthree\n";

        assert_eq!(fit_rows(frame, 80, 24), frame);
    }

    /// fails if the fit counts lines instead of rows. Four lines is four
    /// lines, but at ten columns each of these wraps onto three, so twelve
    /// rows of content do not go into a window of eight.
    #[test]
    fn a_wrapped_line_costs_more_than_one_row() {
        let wide = "0123456789012345678901234";
        let frame = format!("{wide}\n{wide}\n{wide}\n{wide}\n");

        let fitted = fit_rows(&frame, 10, 8);

        assert_eq!(
            fitted, "0123456789012345678901234\n3 more lines than this terminal shows\n",
            "the notice reserves its own four rows at ten columns, leaving three of the seven \
             for content, which is one wrapped line"
        );
    }

    /// fails if the notice steals the row of a line it is reporting, or if
    /// the count goes wrong. Ten lines into a window six rows tall keeps
    /// four and says so.
    #[test]
    fn a_frame_too_tall_says_how_much_it_dropped() {
        let frame = (0..10)
            .map(|n| n.to_string())
            .collect::<Vec<_>>()
            .join("\n");

        let fitted = fit_rows(&frame, 80, 6);

        assert_eq!(
            fitted,
            "0\n1\n2\n3\n6 more lines than this terminal shows\n"
        );
    }

    /// fails if a pty that has never been told its size swallows the flock.
    /// `script(1)` reports zero rows and zero columns, and trimming to that
    /// left nothing on screen but the notice, once a second.
    #[test]
    fn a_terminal_reporting_no_size_gets_the_frame_whole() {
        let frame = "a\nb\nc\n";

        assert_eq!(fit_rows(frame, 0, 0), frame, "no size at all");
        assert_eq!(fit_rows(frame, 80, 0), frame, "no rows");
        assert_eq!(fit_rows(frame, 0, 24), frame, "no columns");
        assert_eq!(
            fit_rows(frame, 80, 1),
            frame,
            "one row leaves nothing to trim to once the cursor has its own"
        );
    }

    /// fails if the host strip stops leading the frame. It has to be first:
    /// it holds still while the tables under it change length, and it is
    /// what survives a window too short for the rest.
    #[test]
    fn a_followed_frame_leads_with_the_host_strip() {
        let usage = HostUsage {
            cpu_percent: Some(11.0),
            memory_used_bytes: 39_963_869_184,
            memory_total_bytes: 51_539_607_552,
            disk_bytes_per_second: Some((0, 0)),
            network_bytes_per_second: Some((0, 0)),
        };
        let strip = host::strip(Some(usage), Presentation::BARE, 200);

        let frame = follow_frame(vec![sample_info()], Presentation::BARE, Some(strip));

        let mut lines = frame.lines();
        assert!(lines.next().unwrap().starts_with("host  cpu  "));
        assert_eq!(lines.next().unwrap(), "");
        assert!(frame.contains("web"), "the table still follows: {frame}");
    }

    /// fails if a shepherd that will not answer for the machine starts
    /// printing a blank strip instead of no strip.
    #[test]
    fn a_frame_without_a_host_strip_is_the_tables_alone() {
        let frame = follow_frame(vec![sample_info()], Presentation::BARE, None);

        assert!(!frame.contains("host  cpu"), "{frame}");
        assert!(frame.contains("web"), "{frame}");
    }

    /// fails if the flourish comes back into a followed frame. It is art
    /// above an empty flock, and a redraw every second turns it into noise.
    #[test]
    fn a_followed_frame_of_an_empty_flock_carries_no_flourish() {
        let frame = follow_frame(Vec::new(), Presentation::BARE, None);

        assert!(!frame.contains("no sheep in the flock yet"), "{frame}");
    }

    /// A reading with every rate present, so a strip built from it names
    /// all four segments.
    fn sample_host() -> HostUsage {
        HostUsage {
            cpu_percent: Some(11.459_433),
            memory_used_bytes: 39_963_869_184,
            memory_total_bytes: 51_539_607_552,
            disk_bytes_per_second: Some((1_258_291, 491_520)),
            network_bytes_per_second: Some((24_594, 9_260)),
        }
    }

    /// A shepherd answering both halves of what a listing asks for, with
    /// `host` as its answer to `Request::HostUsage`.
    ///
    /// The envelope receiver comes back with the client and must be held:
    /// `fake_client_answering`'s loop stops the moment nothing is listening,
    /// which would leave the second request of a listing unanswered.
    async fn shepherd_answering(
        path: &std::path::Path,
        host: Option<HostUsage>,
    ) -> (
        Client,
        tokio::sync::mpsc::UnboundedReceiver<shep_core::protocol::Envelope>,
    ) {
        fake_client_answering(path, move |request| match request {
            Request::HostUsage => Response::HostUsage(host),
            _ => Response::Flock(vec![sample_info()]),
        })
        .await
    }

    /// One `shep flock` rendered into a string, at `fmt`.
    async fn listing(client: &Client, fmt: Format) -> (ExitCode, String) {
        let mut out = Vec::new();
        let mut err = Vec::new();
        let code = {
            let mut streams = Streams {
                out: &mut out,
                err: &mut err,
                style: crate::style::Presentation::BARE,
                fmt,
            };
            flock(client, &mut streams).await
        };
        (code, String::from_utf8(out).unwrap())
    }

    /// The version-skew arm, and the reason a refusal is silent: a shepherd
    /// built before `Request::HostUsage` existed decodes
    /// `Request::Unrecognized` and answers `RpcErrorCode::Unsupported` by
    /// name.
    ///
    /// fails if a refusal starts failing the verb, or starts being told
    /// apart from a reading of `None`, which is a platform answer rather
    /// than a version one.
    #[tokio::test]
    async fn a_shepherd_that_refuses_the_host_request_answers_for_no_strip() {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, _daemon) = fake_client_replying_err(
            &path,
            RpcErrorCode::Unsupported,
            "this shepherd does not implement the request the client sent",
        )
        .await;

        assert_eq!(host_usage(&client).await, None);
    }

    /// fails if a reply this client cannot read starts being rendered as
    /// something. The fake answers `Response::Pong` to everything.
    #[tokio::test]
    async fn a_shepherd_answering_something_else_answers_for_no_strip() {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, _envelopes) = fake_client_capturing_envelopes(&path).await;

        assert_eq!(host_usage(&client).await, None);
    }

    /// fails if a one-shot listing stops asking for the machine. The whole
    /// point of the shepherd holding a baseline is that a listing which
    /// never asks gets nothing for it.
    #[tokio::test]
    async fn flock_asks_for_the_machine_as_well_as_the_flock() {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, mut envelopes) = shepherd_answering(&path, Some(sample_host())).await;

        let _ = listing(&client, Format::Table).await;

        let mut sent = Vec::new();
        for _ in 0..2 {
            let envelope = tokio::time::timeout(RECV_TIMEOUT, envelopes.recv())
                .await
                .expect("flock must reach the wire; it hung instead of sending a request")
                .unwrap();
            sent.push(envelope.body);
        }
        assert_eq!(sent, vec![Request::ListFlock, Request::HostUsage]);
    }

    /// The door an operator actually knocks on. `host::strip` has its own
    /// tests for what the line says; this one is about a one-shot listing
    /// carrying it at all, which is the whole ask.
    ///
    /// fails if the strip stops leading a one-shot table, or stops being
    /// separated from it.
    #[tokio::test]
    async fn a_one_shot_listing_leads_with_the_host_strip() {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, _envelopes) = shepherd_answering(&path, Some(sample_host())).await;

        let (code, text) = listing(&client, Format::Table).await;

        assert_eq!(code, ExitCode::Success);
        let mut lines = text.lines();
        assert!(
            lines.next().unwrap().starts_with("host  cpu  "),
            "got {text:?}"
        );
        assert_eq!(lines.next().unwrap(), "");
        assert!(text.contains("web"), "the table still follows: {text:?}");
    }

    /// fails if the two formats stop answering the same question. The table
    /// shows the machine, so the JSON says so too, beside `data` rather
    /// than inside it: an existing `data[0].name` script sees no change.
    #[tokio::test]
    async fn the_json_surface_carries_the_host_beside_the_flock() {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, _envelopes) = shepherd_answering(&path, Some(sample_host())).await;

        let (_code, text) = listing(&client, Format::Json).await;

        let json: serde_json::Value = serde_json::from_str(&text).unwrap();
        assert_eq!(json["data"][0]["name"], "web");
        assert_eq!(json["host"]["memory_used_bytes"], 39_963_869_184u64);
        assert_eq!(json["host"]["cpu_percent"], 11.459_433);
        // A strip is a terminal thing, and JSON is not a terminal.
        assert!(!text.contains("host  cpu"), "got {text:?}");
    }

    /// Three answers, and a reader has to tell them apart: absent means
    /// this shepherd would not say, `null` means a platform that cannot be
    /// read, and an object with `null` rates means one that has not been
    /// read for long enough yet.
    ///
    /// fails if any two of them collapse onto one spelling.
    #[tokio::test]
    async fn the_json_host_key_spells_its_three_answers_apart() {
        let json = async |host: Option<HostUsage>| {
            let dir = tempfile::tempdir().unwrap();
            let path = shep_client::testing::control_address(dir.path());
            let (client, _envelopes) = shepherd_answering(&path, host).await;
            let (_code, text) = listing(&client, Format::Json).await;
            serde_json::from_str::<serde_json::Value>(&text).unwrap()
        };

        // Read, and read long enough ago to have a window: an object whose
        // rates are numbers.
        let read = json(Some(sample_host())).await;
        assert_eq!(read["host"]["cpu_percent"], 11.459_433);

        // A platform `sysinfo` cannot read: present and null, which is not
        // the same claim as a rate that has no window yet.
        let unreadable = json(None).await;
        assert!(unreadable["host"].is_null(), "got {unreadable:?}");

        // Read, but with no window yet: an object, with nulls where the
        // rates will be.
        let unmeasured = json(Some(HostUsage {
            cpu_percent: None,
            disk_bytes_per_second: None,
            network_bytes_per_second: None,
            ..sample_host()
        }))
        .await;
        assert!(unmeasured["host"].is_object(), "got {unmeasured:?}");
        assert!(unmeasured["host"]["cpu_percent"].is_null());
        assert_eq!(unmeasured["host"]["memory_used_bytes"], 39_963_869_184u64);
    }

    /// The third spelling: no key at all, from a shepherd that would not
    /// answer the question. A script has to be able to tell "this shepherd
    /// is too old" from "this machine cannot be read".
    ///
    /// fails if a refusal starts writing `host: null`, which would claim a
    /// platform answer the shepherd never gave.
    #[tokio::test]
    async fn a_refused_host_request_leaves_no_json_key_at_all() {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, daemon) = fake_client_with_ack(&path, sample_ack()).await;
        daemon.reply_to_list(vec![sample_info()]);

        // `FakeDaemon` answers anything it has no script for with
        // `Response::Pong`, which this client can no more use than a
        // refusal.
        let (code, text) = listing(&client, Format::Json).await;

        assert_eq!(code, ExitCode::Success);
        let json: serde_json::Value = serde_json::from_str(&text).unwrap();
        assert_eq!(json["data"][0]["name"], "web");
        assert!(json.get("host").is_none(), "got {json:?}");
    }

    #[tokio::test]
    async fn flock_asks_the_daemon_to_list_the_whole_flock() {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, mut envelopes) = fake_client_capturing_envelopes(&path).await;
        let mut out = Vec::new();
        let mut err = Vec::new();
        let mut streams = Streams {
            out: &mut out,
            err: &mut err,
            style: crate::style::Presentation::BARE,
            fmt: Format::Table,
        };
        let _ = flock(&client, &mut streams).await;
        let sent = tokio::time::timeout(RECV_TIMEOUT, envelopes.recv())
            .await
            .expect("flock must reach the wire; it hung instead of sending a request")
            .unwrap();
        assert_eq!(sent.body, Request::ListFlock);
    }

    #[tokio::test]
    async fn describe_sends_the_parsed_selector_in_its_compiled_form() {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, mut envelopes) = fake_client_capturing_envelopes(&path).await;
        let paths = ShepPaths::resolve(&|_| None, dir.path());

        for (input, expected) in [
            ("all", SelectorSpec::All),
            ("7", SelectorSpec::Id(7)),
            ("web", SelectorSpec::Name("web".into())),
            ("/^web-/", SelectorSpec::Regex("^web-".into())),
            ("fold:api", SelectorSpec::Fold("api".into())),
        ] {
            let mut out = Vec::new();
            let mut err = Vec::new();
            let mut streams = Streams {
                out: &mut out,
                err: &mut err,
                style: crate::style::Presentation::BARE,
                fmt: Format::Table,
            };
            let args = SelectorArgs {
                selectors: vec![input.into()],
            };
            let _ = describe(&client, &mut streams, &paths, &args).await;
            let sent = tokio::time::timeout(RECV_TIMEOUT, envelopes.recv())
                .await
                .unwrap_or_else(|_| {
                    panic!("describe({input}) must reach the wire; it hung instead of sending a request")
                })
                .unwrap();
            assert_eq!(
                sent.body,
                Request::Describe { selector: expected },
                "{input}"
            );
        }
    }

    /// `"/[/"` is one of the only three inputs the selector grammar rejects:
    /// an unterminated regex character class.
    #[tokio::test]
    async fn a_malformed_selector_exits_usage_without_a_round_trip() {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, mut envelopes) = fake_client_capturing_envelopes(&path).await;
        let paths = ShepPaths::resolve(&|_| None, dir.path());
        let mut out = Vec::new();
        let mut err = Vec::new();
        let code = {
            let mut streams = Streams {
                out: &mut out,
                err: &mut err,
                style: crate::style::Presentation::BARE,
                fmt: Format::Table,
            };
            describe(
                &client,
                &mut streams,
                &paths,
                &SelectorArgs {
                    selectors: vec!["/[/".into()],
                },
            )
            .await
        };
        assert_eq!(code, ExitCode::Usage);
        assert!(
            envelopes.try_recv().is_err(),
            "a malformed selector must fail locally"
        );
    }

    #[tokio::test]
    async fn fold_asks_the_daemon_for_that_fold_and_nothing_wider() {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, mut envelopes) = fake_client_capturing_envelopes(&path).await;
        let paths = ShepPaths::resolve(&|_| None, dir.path());
        let mut out = Vec::new();
        let mut err = Vec::new();
        let mut streams = Streams {
            out: &mut out,
            err: &mut err,
            style: crate::style::Presentation::BARE,
            fmt: Format::Table,
        };
        let _ = fold(
            &client,
            &mut streams,
            &paths,
            &FoldArgs { name: "api".into() },
        )
        .await;
        let sent = tokio::time::timeout(RECV_TIMEOUT, envelopes.recv())
            .await
            .expect("fold must reach the wire; it hung instead of sending a request")
            .unwrap();
        assert_eq!(
            sent.body,
            Request::Describe {
                selector: SelectorSpec::Fold("api".into())
            }
        );
    }

    /// `Response::Flock` and `Response::Described` both wrap a bare
    /// `Vec<ProcessInfo>`, so an arm swapped between them compiles clean.
    /// `reply_to_list` scripts a real `Response::Flock` to catch that.
    #[tokio::test]
    async fn flock_response_round_trips_into_rendered_flock_rows() {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, daemon) = fake_client_with_ack(&path, sample_ack()).await;
        daemon.reply_to_list(vec![sample_info()]);

        let mut out = Vec::new();
        let mut err = Vec::new();
        let code = {
            let mut streams = Streams {
                out: &mut out,
                err: &mut err,
                style: crate::style::Presentation::BARE,
                fmt: Format::Json,
            };
            flock(&client, &mut streams).await
        };

        assert_eq!(code, ExitCode::Success);
        let json: serde_json::Value = serde_json::from_slice(&out).unwrap();
        assert_eq!(json["command"], "flock");
        assert_eq!(json["data"][0]["name"], "web");
    }

    #[tokio::test]
    async fn describe_response_round_trips_into_rendered_flock_rows() {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, daemon) = fake_client_with_ack(&path, sample_ack()).await;
        daemon.reply_to_describe(vec![sample_info()]);
        let paths = ShepPaths::resolve(&|_| None, dir.path());

        let mut out = Vec::new();
        let mut err = Vec::new();
        let code = {
            let mut streams = Streams {
                out: &mut out,
                err: &mut err,
                style: crate::style::Presentation::BARE,
                fmt: Format::Json,
            };
            describe(
                &client,
                &mut streams,
                &paths,
                &SelectorArgs {
                    selectors: vec!["all".into()],
                },
            )
            .await
        };

        assert_eq!(code, ExitCode::Success);
        let json: serde_json::Value = serde_json::from_slice(&out).unwrap();
        assert_eq!(json["command"], "describe");
        assert_eq!(json["data"][0]["name"], "web");
    }

    #[test]
    fn describe_lists_secret_references_with_a_verdict_and_no_values() {
        let rendered = render_describe_secrets(&[
            ("DB_PASSWORD", "production", Resolution::Found("hunter2")),
            ("vercel/API_KEY", "production", Resolution::MissingNamespace),
            ("ABSENT", "production", Resolution::MissingKey),
        ]);
        assert!(rendered.contains("DB_PASSWORD"));
        assert!(rendered.contains("vercel/API_KEY"));
        assert!(!rendered.contains("hunter2"), "never a value");
    }

    /// A roll entry, an operator store entry and no provider cache at all:
    /// the three verdicts `gather_secrets` can produce from local files
    /// alone, exercised through the real `describe` verb rather than
    /// `render_describe_secrets` in isolation.
    async fn describe_with_a_seeded_web(
        fmt: Format,
    ) -> (ExitCode, shep_core::paths::ShepPaths, Vec<u8>) {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, daemon) = fake_client_with_ack(&path, sample_ack()).await;
        daemon.reply_to_describe(vec![
            ProcessInfo::builder(1, "web", ProcStatus::Online).build(),
        ]);

        // `SHEP_HOME` pinned to `dir` itself, not its `.shep` default: this
        // test writes `paths.snapshot`/`paths.secrets` directly, and the
        // default subdirectory is never created outside a real boot.
        let home = dir.path().display().to_string();
        let paths = ShepPaths::resolve(
            &move |key| (key == "SHEP_HOME").then(|| home.clone()),
            dir.path(),
        );

        let mut config = shep_core::config::AppConfig::minimal("web", "./srv");
        config
            .env
            .insert("A".into(), "{{secret:DB_PASSWORD}}".into());
        config
            .env
            .insert("B".into(), "{{secret:vercel/API_KEY}}".into());
        let roll = FlockSnapshot {
            version: 1,
            saved_at_ms: 0,
            apps: vec![shep_daemon::snapshot::SavedApp {
                app: config,
                instances_running: 1,
            }],
        };
        std::fs::write(&paths.snapshot, serde_json::to_vec(&roll).unwrap()).unwrap();
        secrets::set(&paths.secrets, "DB_PASSWORD", "production", "hunter2").unwrap();
        // No provider cache written: `vercel` has never pushed anything.

        let mut out = Vec::new();
        let mut err = Vec::new();
        let code = {
            let mut streams = Streams {
                out: &mut out,
                err: &mut err,
                style: crate::style::Presentation::BARE,
                fmt,
            };
            describe(
                &client,
                &mut streams,
                &paths,
                &SelectorArgs {
                    selectors: vec!["all".into()],
                },
            )
            .await
        };
        (code, paths, out)
    }

    #[tokio::test]
    async fn describe_prints_real_secret_verdicts_in_the_table() {
        let (code, _paths, out) = describe_with_a_seeded_web(Format::Table).await;
        assert_eq!(code, ExitCode::Success);
        let rendered = String::from_utf8(out).unwrap();
        // `emit_described` places this block once, right after Overridden.
        // A second renderer writing the same block again is the exact bug
        // this pins: `.contains()` alone is true whether it printed once
        // or twice.
        assert_eq!(rendered.matches("Secrets for web").count(), 1, "{rendered}");
        assert!(
            rendered.contains("DB_PASSWORD (production): resolved"),
            "{rendered}"
        );
        assert!(
            rendered
                .contains("vercel/API_KEY (production): not cached; a provider may still have it"),
            "{rendered}"
        );
        assert!(!rendered.contains("hunter2"), "never a value: {rendered}");
    }

    #[tokio::test]
    async fn describe_json_carries_the_same_verdicts_as_an_additive_field() {
        let (code, _paths, out) = describe_with_a_seeded_web(Format::Json).await;
        assert_eq!(code, ExitCode::Success);
        let json: serde_json::Value = serde_json::from_slice(&out).unwrap();
        assert_eq!(json["command"], "describe");
        assert_eq!(json["schema_version"], 1, "SCHEMA_VERSION must not move");
        // `data` stays exactly what it always was: an array of ProcessInfo.
        assert_eq!(json["data"][0]["name"], "web");
        let entries = json["secrets"].as_array().unwrap();
        assert_eq!(entries.len(), 2);
        assert!(
            entries
                .iter()
                .any(|e| e["reference"] == "DB_PASSWORD" && e["status"] == "resolved"),
            "{entries:?}"
        );
        assert!(
            entries
                .iter()
                .any(|e| e["reference"] == "vercel/API_KEY" && e["status"] == "uncached"),
            "{entries:?}"
        );
        assert!(!out_contains(&json, "hunter2"), "never a value");
    }

    /// The CLI reads the same pairs the shepherd resolves against, so a
    /// namespace pushed for `production` reads as uncached for a `staging`
    /// sheep rather than as a key the provider does not have. Reporting
    /// `missing` here would tell an operator to go and find a value that a
    /// dog mid-poll is about to supply.
    #[tokio::test]
    async fn describe_reads_a_namespace_pushed_for_another_environment_as_uncached() {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, daemon) = fake_client_with_ack(&path, sample_ack()).await;
        daemon.reply_to_describe(vec![
            ProcessInfo::builder(1, "web", ProcStatus::Online).build(),
        ]);
        let home = dir.path().display().to_string();
        let paths = ShepPaths::resolve(
            &move |key| (key == "SHEP_HOME").then(|| home.clone()),
            dir.path(),
        );

        let mut config = shep_core::config::AppConfig::minimal("web", "./srv");
        config.environment = Some("staging".into());
        config
            .env
            .insert("B".into(), "{{secret:vercel/API_KEY}}".into());
        let roll = FlockSnapshot {
            version: 1,
            saved_at_ms: 0,
            apps: vec![shep_daemon::snapshot::SavedApp {
                app: config,
                instances_running: 1,
            }],
        };
        std::fs::write(&paths.snapshot, serde_json::to_vec(&roll).unwrap()).unwrap();
        // The pair the dog has pushed is production; staging is still to come.
        std::fs::write(
            &paths.secrets_cache,
            r#"{"version":2,"namespaces":{"vercel":{"API_KEY":{"production":"sk_live"}}},"pushed":{"vercel":["production"]}}"#,
        )
        .unwrap();

        let mut out = Vec::new();
        let mut err = Vec::new();
        let code = {
            let mut streams = Streams {
                out: &mut out,
                err: &mut err,
                style: crate::style::Presentation::BARE,
                fmt: Format::Json,
            };
            describe(
                &client,
                &mut streams,
                &paths,
                &SelectorArgs {
                    selectors: vec!["all".into()],
                },
            )
            .await
        };

        assert_eq!(code, ExitCode::Success);
        let json: serde_json::Value = serde_json::from_slice(&out).unwrap();
        let entries = json["secrets"].as_array().unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0]["environment"], "staging");
        assert_eq!(entries[0]["status"], "uncached", "{entries:?}");
        assert!(!out_contains(&json, "sk_live"), "never a value");
    }

    /// A store that will not parse is not a store with nothing in it. The
    /// verdicts still say `missing`, since nothing local can say otherwise,
    /// but the operator is told why before being sent to `shep secret set`
    /// for a value the store may already hold.
    #[tokio::test]
    async fn describe_says_the_store_would_not_read_rather_than_calling_keys_missing() {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, daemon) = fake_client_with_ack(&path, sample_ack()).await;
        daemon.reply_to_describe(vec![
            ProcessInfo::builder(1, "web", ProcStatus::Online).build(),
        ]);
        let home = dir.path().display().to_string();
        let paths = ShepPaths::resolve(
            &move |key| (key == "SHEP_HOME").then(|| home.clone()),
            dir.path(),
        );

        let mut config = shep_core::config::AppConfig::minimal("web", "./srv");
        config
            .env
            .insert("A".into(), "{{secret:DB_PASSWORD}}".into());
        let roll = FlockSnapshot {
            version: 1,
            saved_at_ms: 0,
            apps: vec![shep_daemon::snapshot::SavedApp {
                app: config,
                instances_running: 1,
            }],
        };
        std::fs::write(&paths.snapshot, serde_json::to_vec(&roll).unwrap()).unwrap();
        std::fs::write(&paths.secrets, b"{not json").unwrap();

        let mut out = Vec::new();
        let mut err = Vec::new();
        let code = {
            let mut streams = Streams {
                out: &mut out,
                err: &mut err,
                style: crate::style::Presentation::BARE,
                fmt: Format::Table,
            };
            describe(
                &client,
                &mut streams,
                &paths,
                &SelectorArgs {
                    selectors: vec!["all".into()],
                },
            )
            .await
        };

        assert_eq!(code, ExitCode::Success);
        let warning = String::from_utf8(err).unwrap();
        assert!(
            warning.contains("could not be read"),
            "the notice has to say the store failed: {warning}"
        );
        assert!(
            warning.contains(&paths.secrets.display().to_string()),
            "and name the file: {warning}"
        );
        let rendered = String::from_utf8(out).unwrap();
        assert!(rendered.contains("DB_PASSWORD"), "{rendered}");
    }

    /// Whether `hunter2` shows up anywhere in `value`'s own JSON text, not
    /// just at the top level: a value smuggled in nested one level deeper
    /// would still be a leak.
    fn out_contains(value: &serde_json::Value, needle: &str) -> bool {
        value.to_string().contains(needle)
    }

    /// `fold` shares `describe_selector` but must stay byte-identical to
    /// before this feature existed: no local file I/O, no `secrets` field.
    #[tokio::test]
    async fn fold_never_computes_or_prints_a_secrets_section() {
        let dir = tempfile::tempdir().unwrap();
        let path = shep_client::testing::control_address(dir.path());
        let (client, daemon) = fake_client_with_ack(&path, sample_ack()).await;
        daemon.reply_to_describe(vec![
            ProcessInfo::builder(1, "web", ProcStatus::Online).build(),
        ]);
        // Deliberately no `paths.snapshot` on disk: if `fold` ever reads it,
        // the missing file is tolerated (`read_roll`), which would hide the
        // bug this test exists to catch. The real guard is the assertion
        // below, on `command == "fold"` never entering `gather_secrets`.
        let paths = ShepPaths::resolve(&|_| None, dir.path());

        let mut out = Vec::new();
        let mut err = Vec::new();
        let mut streams = Streams {
            out: &mut out,
            err: &mut err,
            style: crate::style::Presentation::BARE,
            fmt: Format::Json,
        };
        let code = fold(
            &client,
            &mut streams,
            &paths,
            &FoldArgs { name: "api".into() },
        )
        .await;
        assert_eq!(code, ExitCode::Success);
        let json: serde_json::Value = serde_json::from_slice(&out).unwrap();
        assert!(json.get("secrets").is_none(), "{json}");
    }

    // --- sheep_flourish ---

    /// A sheep with `status` pinned and nothing else.
    fn sheep(id: u32, status: ProcStatus) -> ProcessInfo {
        ProcessInfo::builder(id, format!("s{id}"), status).build()
    }

    /// A registered dog, which `sheep_flourish` must never count as a sheep.
    fn dog(id: u32) -> ProcessInfo {
        ProcessInfo::builder(id, format!("d{id}"), ProcStatus::Online)
            .dog(Some(DogSource::BuiltIn))
            .build()
    }

    #[test]
    fn sheep_flourish_fires_empty_flock_on_a_truly_empty_listing() {
        let art = sheep_flourish(&[]).expect("an empty listing must flourish");
        assert!(art.contains("no sheep in the flock yet"), "{art}");
    }

    #[test]
    fn sheep_flourish_treats_dogs_only_as_an_empty_flock() {
        let art =
            sheep_flourish(&[dog(1), dog(2)]).expect("dogs alone must read as an empty flock");
        assert!(art.contains("no sheep in the flock yet"), "{art}");
    }

    /// The count in the flourish excludes dogs too.
    #[test]
    fn sheep_flourish_fires_all_asleep_when_every_sheep_is_stopped() {
        let listing = [
            sheep(1, ProcStatus::Stopped),
            sheep(2, ProcStatus::Stopped),
            dog(3),
        ];
        let art = sheep_flourish(&listing).expect("an all-stopped flock must flourish");
        assert!(art.contains("2 in the flock, all asleep"), "{art}");
    }

    #[test]
    fn a_live_dog_does_not_block_all_asleep() {
        let listing = [sheep(1, ProcStatus::Stopped), dog(2)];
        let art = sheep_flourish(&listing).expect("a live dog must not suppress all_asleep");
        assert!(art.contains("1 in the flock, all asleep"), "{art}");
    }

    #[test]
    fn sheep_flourish_is_silent_on_a_mixed_flock() {
        let listing = [sheep(1, ProcStatus::Online), sheep(2, ProcStatus::Stopped)];
        assert_eq!(
            sheep_flourish(&listing),
            None,
            "a mixed flock is not a flourish moment"
        );
    }

    /// `Stopping` is reload's transient for the instance being replaced, so
    /// a flock mid-reload must not read as asleep.
    #[test]
    fn stopping_does_not_count_as_asleep() {
        let listing = [
            sheep(1, ProcStatus::Stopping),
            sheep(2, ProcStatus::Stopping),
        ];
        assert_eq!(
            sheep_flourish(&listing),
            None,
            "Stopping is a transient, not rest"
        );
    }

    /// The daemon answers an empty flock on every case here, which
    /// `sheep_flourish` always fires on, so only the gate decides.
    #[tokio::test]
    async fn the_flourish_only_prints_under_table_format_and_a_sheep_drawing_level() {
        use crate::style::{Presentation, StyleLevel};

        for (fmt, level, expect_art) in [
            (Format::Table, StyleLevel::Full, true),
            (Format::Json, StyleLevel::Full, false),
            (Format::Table, StyleLevel::Plain, false),
            (Format::Table, StyleLevel::Bare, false),
        ] {
            let dir = tempfile::tempdir().unwrap();
            let path = shep_client::testing::control_address(dir.path());
            let (client, _daemon) = fake_client_on(&path).await;

            let mut out = Vec::new();
            let mut err = Vec::new();
            let mut streams = Streams {
                out: &mut out,
                err: &mut err,
                style: Presentation::new(level, None, None, None, 80),
                fmt,
            };
            let _ = flock(&client, &mut streams).await;
            let printed = String::from_utf8_lossy(&out);
            assert_eq!(
                printed.contains("no sheep in the flock yet"),
                expect_art,
                "fmt={fmt:?} level={level:?}: {printed}"
            );
        }
    }
}