shep_core/protocol/

request.rs

//! RPC frames: requests, responses, envelopes, and structured errors

use core::fmt;

use std::collections::BTreeMap;

use serde::{Deserialize, Deserializer, Serialize};

use crate::config::{AppConfig, DeclaredApp, ResetDepth};
use crate::status::ProcStatus;

/// Client's opening frame
///
/// No `deny_unknown_fields`: refusing an unknown field here would refuse a
/// newer client before `protocol` is read.
// wire format: changing this is a breaking change
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Hello {
    /// Client crate version (semver string)
    pub client_version: String,
    /// [`crate::protocol::PROTOCOL_VERSION`] the client speaks
    pub protocol: u32,
    /// The name this client was registered under as a dog, when it is one.
    ///
    /// `None` for every other client; a bare `Client` cannot set it. The
    /// daemon needs it to name a dog it refuses at the handshake, which never
    /// reaches `Request::DogConfig`. A dog reads its own name from
    /// `$SHEP_DOG_NAME`.
    ///
    /// Absent on the wire rather than `null`, so
    /// [`crate::protocol::PROTOCOL_VERSION`] does not move for it.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub dog_name: Option<String>,
}

/// Daemon's handshake answer
// wire format: changing this is a breaking change
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct HelloAck {
    /// Daemon crate version
    pub daemon_version: String,
    /// Protocol version the daemon speaks
    pub protocol: u32,
    /// Daemon pid
    pub pid: u32,
    /// The oldest protocol this daemon accepts, or `None` from a daemon
    /// predating the floor.
    ///
    /// Absent rather than `null` on the wire, so it does not move
    /// [`crate::protocol::PROTOCOL_VERSION`].
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub min_supported: Option<u32>,
}

/// Serializable selector (mirror of [`crate::selector::ProcessSelector`];
/// regex travels as its source string)
// wire format: changing this is a breaking change
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "value", rename_all = "snake_case")]
pub enum SelectorSpec {
    /// Every sheep
    All,
    /// By id
    Id(u32),
    /// By exact name
    Name(String),
    /// By regex source
    Regex(String),
    /// By fold name
    Fold(String),
    // Both field names are wire contract, pinned by `request_wire_v8`.
    /// By app name and instance slot
    ///
    /// On the wire: `{"kind":"instance","value":{"name":"web","slot":2}}`.
    Instance {
        /// The app name
        name: String,
        /// The instance slot, counting from 0
        slot: u32,
    },
}

/// A short marker a dog attaches to a sheep for `shep flock` to paint.
///
/// shep stores and prints it, never parses it: `▲ main@a1b2c3` is a
/// deploy tool's sentence.
///
/// The grammar: non-empty once whitespace is discounted, at most
/// [`Self::MAX_CHARS`] characters, no [`char::is_control`] character,
/// `\u{1b}` included. Refused, never repaired, and validated here rather
/// than at the renderer: `shep`'s own `output::width::sanitize_cell` keeps
/// a well-formed CSI sequence, since shep's colouring is made of them.
///
/// [`Self::MAX_CHARS`] counts `char`s, not bytes: a byte cap would refuse a
/// legitimate CJK smit at roughly a third of its apparent length.
///
/// `Debug` is derived: a smit carries no secret, so there is nothing to
/// redact.
///
/// # Example
/// ```
/// use shep_core::protocol::Smit;
///
/// assert_eq!("▲ main@a1b2c3".parse::<Smit>()?.as_str(), "▲ main@a1b2c3");
/// assert!("\u{1b}[2Jgone".parse::<Smit>().is_err()); // no escapes
/// # Ok::<(), shep_core::protocol::SmitError>(())
/// ```
// wire format: changing this is a breaking change
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct Smit(String);

impl Smit {
    /// The longest a smit may be, in characters.
    pub const MAX_CHARS: usize = 48;

    /// The marker as text, exactly as its publisher sent it.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

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

impl core::str::FromStr for Smit {
    type Err = SmitError;

    /// # Errors
    /// - [`SmitError::Empty`] if the text is nothing but whitespace.
    /// - [`SmitError::TooLong`] if it is over [`Self::MAX_CHARS`] characters.
    /// - [`SmitError::Unprintable`] if it holds a control character.
    fn from_str(text: &str) -> Result<Self, Self::Err> {
        if text.trim().is_empty() {
            return Err(SmitError::Empty);
        }
        let chars = text.chars().count();
        if chars > Self::MAX_CHARS {
            return Err(SmitError::TooLong { chars });
        }
        if text.chars().any(char::is_control) {
            return Err(SmitError::Unprintable);
        }
        Ok(Self(text.to_string()))
    }
}

/// Validates on decode: a dog written in another language speaks this wire
/// directly and never runs [`core::str::FromStr`], so a derived impl would
/// let `\u{1b}[2J` reach every listing built from a smit.
impl<'de> Deserialize<'de> for Smit {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        // String, not &str: a non-borrowing deserializer cannot always borrow
        let text = String::deserialize(deserializer)?;
        text.parse().map_err(serde::de::Error::custom)
    }
}

/// Why a string is not a [`Smit`].
#[non_exhaustive]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SmitError {
    /// Over [`Smit::MAX_CHARS`] characters; carries the count that was sent.
    TooLong {
        /// How many characters the string held.
        chars: usize,
    },
    /// A control character, `\u{1b}` included.
    Unprintable,
    /// Empty, or nothing but whitespace.
    Empty,
}

impl fmt::Display for SmitError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::TooLong { chars } => write!(
                f,
                "a smit is at most {} characters; this one is {chars}",
                Smit::MAX_CHARS
            ),
            Self::Unprintable => {
                f.write_str("a smit may not contain a control character, an escape included")
            }
            Self::Empty => f.write_str("a smit may not be empty"),
        }
    }
}

impl core::error::Error for SmitError {}

/// One RPC request
// wire format: changing existing variants is a breaking change
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum Request {
    /// Liveness check
    Ping,
    /// Full flock listing
    ListFlock,
    /// Detailed info for matching sheep
    Describe {
        /// Which sheep
        selector: SelectorSpec,
    },
    /// Register + start apps
    Start {
        /// App configs. The daemon must re-normalize them, since peer input
        /// is untrusted; failures return [`RpcErrorCode::InvalidConfig`]
        apps: Vec<AppConfig>,
    },
    /// Register apps as flock members without starting any of them
    ///
    /// Each app lands `Stopped` and holds no pid; `shep add` is the verb.
    ///
    /// Idempotent by name: an app the flock already has is answered as it
    /// stands, running or not, and nothing about it changes.
    /// [`Self::ApplyConfig`] merges a template into one the flock already
    /// has, and `shep add` sends both.
    ///
    /// Answers [`Response::Added`].
    Add {
        /// App configs, carried exactly as [`Self::Start`] carries them. The
        /// daemon must re-normalize them, since peer input is untrusted;
        /// failures return [`RpcErrorCode::InvalidConfig`]
        apps: Vec<AppConfig>,
    },
    /// Ask which of `apps` name a sheep the flock already has under a
    /// different config
    ///
    /// Read-only. [`Self::Start`] on an already-registered name adds
    /// instances rather than reconciling config.
    ///
    /// Answers [`Response::Drifted`] with one [`SheepDrift`] per app that is
    /// both registered and different. An app the flock does not have is
    /// absent from the answer, not reported as unchanged.
    ConfigDrift {
        /// The configs to compare against, exactly as [`Self::Start`] would
        /// carry them. The daemon must re-normalize them: peer input is
        /// untrusted, and an unnormalized config would report every default
        /// it has not spelled out as a difference. Failures return
        /// [`RpcErrorCode::InvalidConfig`].
        apps: Vec<AppConfig>,
    },
    /// Merge each declared app into the sheep of the same name, applying
    /// what can be applied and parking the rest for that sheep's next spawn
    ///
    /// Nothing is registered, nothing is pruned and nothing running is
    /// killed: an app the flock does not have is refused by name, and a
    /// field the running child was spawned from waits for a `shep reload`.
    /// Additive by default; `reset` widens it.
    ///
    /// Answers [`Response::Applied`] with one [`SheepApplied`] per entry in
    /// `apps`, in the order given, found or not and changed or not. One
    /// app's refusal rides in [`SheepApplied::refused`] and does not cost
    /// the rest of the file its load.
    ApplyConfig {
        /// The apps to merge in, each carrying the keys its document
        /// literally wrote. The daemon must re-normalize the merge result,
        /// since peer input is untrusted, and refuses the whole request with
        /// [`RpcErrorCode::InvalidConfig`] when two entries share a name:
        /// the second would be merged against a store the first has not
        /// written yet.
        apps: Vec<DeclaredApp>,
        /// How much of what the operator has set since a template last
        /// loaded this request may overwrite. Default [`ResetDepth::None`],
        /// which overwrites nothing.
        ///
        /// Spelled `none`/`file`/`env`/`policy` on the wire.
        reset: ResetDepth,
    },
    /// One sheep's effective config, for a pane that is about to edit it.
    ///
    /// `env` comes back emptied and its key names ride separately, so a
    /// value never crosses the wire. Read-only: nothing about the sheep
    /// changes.
    ///
    /// Answers [`Response::SheepConfig`], or
    /// [`RpcErrorCode::NotFound`] when no sheep has that name.
    SheepConfig {
        /// The sheep's name, not a selector: a pane edits one sheep, for
        /// the reason [`Self::Scale`] states at length.
        name: String,
    },
    /// Sets, replaces, or with `None` removes one env key on one sheep,
    /// recorded as an operator override. Never reads it back.
    ///
    /// Its own request rather than a [`Self::ApplyConfig`] depth, because
    /// no depth does this: `ResetDepth::None` appends only, `File` and
    /// `Policy` leave env alone, and `Env`/`All` replace the whole map with
    /// the template's. A pane cannot send the whole map, since it is never
    /// told the values it would have to send back.
    ///
    /// The running child holds the env it was spawned from, so the change
    /// parks for the next spawn exactly as `ApplyConfig` parks a
    /// respawn-only field, and `shep reload`/`shep restart` promote it.
    ///
    /// Answers [`Response::SheepEnvSet`], or
    /// [`RpcErrorCode::NotFound`] when no sheep has that name.
    SetSheepEnv {
        /// The sheep's name, not a selector, for [`Self::SheepConfig`]'s
        /// reason.
        name: String,
        /// The env key.
        key: String,
        /// The value, or `None` to remove the key.
        ///
        /// [`EnvValue`], not a bare `String`, for the reason that type's
        /// own doc gives: this is the most secret-dense field on the wire
        /// and a derived `Debug` on [`Request`] would print it (IR-41).
        value: Option<EnvValue>,
    },
    /// Sets several env keys on one sheep in a single write.
    ///
    /// [`Self::SetSheepEnv`]'s doc says a request taking a map would need
    /// per-key reporting back "for no caller that wants it". `shep import
    /// env` is that caller: it writes twenty keys at once and has to refuse
    /// the whole set rather than leave eleven of them applied.
    ///
    /// The daemon applies this as one read-modify-write of the override
    /// store, so every key lands or none does. No removal arm: a pane
    /// deletes rows one at a time through [`Self::SetSheepEnv`], and an
    /// import never removes anything.
    ///
    /// A key already holding a different value is a collision. Without
    /// `force` any collision refuses the whole request and writes nothing;
    /// with it, the collisions are overwritten and named in the reply as
    /// well as counted in `set`. A key already holding the same value is
    /// `unchanged`, never a collision, so re-running an unchanged import
    /// needs no flag.
    ///
    /// `dry_run` computes the three lists and writes nothing.
    ///
    /// Parks for the next spawn, exactly as [`Self::SetSheepEnv`] does.
    ///
    /// Answers [`Response::SheepEnvBatch`], or
    /// [`RpcErrorCode::NotFound`] when no sheep has that name.
    SetSheepEnvBatch {
        /// The sheep's name.
        name: String,
        /// The keys and their values.
        ///
        /// [`EnvValue`], not `String`, for [`Self::SetSheepEnv`]'s reason:
        /// `Request` derives `Debug` and this map is the densest run of
        /// secrets on the wire (IR-41).
        entries: BTreeMap<String, EnvValue>,
        /// Overwrite colliding keys instead of refusing.
        force: bool,
        /// Report what would happen and write nothing.
        dry_run: bool,
    },
    /// Sets one config field on one sheep, recorded as an operator
    /// override.
    ///
    /// [`Self::SetSheepEnv`]'s twin for everything that is not `env`, and
    /// it exists for the reason that one does rather than by symmetry.
    /// [`Self::ApplyConfig`] can move a single field (one [`DeclaredApp`]
    /// declaring one key, at [`ResetDepth::File`]), but it moves it as a
    /// template and spends the operator's override for it. That reasoning
    /// does not hold here: a pane's value is the operator's, and the sheep
    /// still differs from its file. Routed through `ApplyConfig`, the `*`
    /// marker would never appear for that edit.
    ///
    /// One field, not a map: a pane edits one row at a time, and a request
    /// that took several would need [`Response::Applied`]'s per-field
    /// reporting back again for no caller that wants it.
    ///
    /// `env` is refused here and goes through [`Self::SetSheepEnv`]. So are
    /// `name` and `instances`, which are
    /// [`ApplyGroup::Structural`](crate::config::ApplyGroup::Structural):
    /// identity and flock shape rather than runtime knobs, and the count
    /// moves through [`Self::Scale`].
    ///
    /// The four-way apply classification governs exactly as it does for a
    /// load. A `Live` field is in force at the daemon's next decision, a
    /// `NextSpawn` field reaches the stored spec, and a `NeedsRespawn`
    /// field parks for `shep reload` to promote.
    ///
    /// Answers [`Response::SheepFieldSet`], or
    /// [`RpcErrorCode::NotFound`] when no sheep has that name.
    SetSheepField {
        /// The sheep's name, not a selector, for [`Self::SheepConfig`]'s
        /// reason.
        name: String,
        /// The [`AppConfig`] field to set. A key that type has no such
        /// field is refused with [`RpcErrorCode::InvalidConfig`] rather
        /// than ignored.
        key: String,
        /// The new value, in the shape that field serializes as. The daemon
        /// must re-validate the resulting config (peer input is untrusted)
        /// and refuses with [`RpcErrorCode::InvalidConfig`] when it does not
        /// deserialize or does not normalize; nothing is written in either
        /// case.
        ///
        /// A bare [`serde_json::Value`] and not a redacting newtype, unlike
        /// [`Self::SetSheepEnv`]'s [`EnvValue`], and the asymmetry is
        /// deliberate. `env` is the one field [`AppConfig`]'s own manual
        /// `Debug` redacts; `cwd`, `script` and `args` are printed in the
        /// clear by every request that already carries a whole config
        /// ([`Self::Start`], [`Self::Add`], [`Self::ApplyConfig`]). A
        /// newtype here would protect one copy of a value this enum prints
        /// three other ways, which reads as a guarantee the wire does not
        /// make. Widening that protection is a change to [`AppConfig`]'s
        /// `Debug`, not to this field.
        value: serde_json::Value,
    },
    /// Replaces one dog's `[<name>]` section in `dogs.toml` and publishes
    /// `config.dog.<name>` so a running dog re-reads it.
    ///
    /// The writing twin of [`Self::DogConfig`], which reads the same
    /// section.
    ///
    /// Answers [`Response::DogConfigSet`].
    SetDogConfig {
        /// The dog's name, the config key.
        name: String,
        /// The whole section, as TOML text.
        ///
        /// [`DogSectionToml`], not a bare `String`, for the reason that
        /// type's own doc gives: a section can hold a dog's credentials and
        /// this is what keeps them out of a `{:?}` (IR-41).
        toml: DogSectionToml,
    },
    /// A provider dog's values for one namespace and one environment.
    ///
    /// Replaces that pair rather than merging into it, so a key deleted at
    /// the provider disappears here on the next push instead of lingering.
    ///
    /// `namespace` is the dog's own registered name. It is bookkeeping, not
    /// authorization: `Hello::dog_name` is self-declared and nothing checks
    /// it against the spawn. The boundary is the socket itself, which lives
    /// under `$SHEP_HOME` at `0700`.
    ///
    /// The two names and every entry key are checked against
    /// [`crate::secrets::is_name`], and a value against
    /// [`crate::secrets::MAX_VALUE_BYTES`], the same cap the operator's own
    /// store enforces. One offender refuses the whole push rather than
    /// dropping its own entry, so a dog never reads `accepted` for a set
    /// that was stored in part.
    ///
    /// Answers [`Response::SecretsPut`].
    PutSecrets {
        /// The dog's registered name.
        namespace: String,
        /// Which environment these values are for.
        environment: String,
        /// The values, keyed by secret name. [`EnvValue`] so a `{:?}` of
        /// this request cannot print them.
        entries: BTreeMap<String, EnvValue>,
    },
    /// Stop matching sheep (stay registered)
    Stop {
        /// Which sheep
        selector: SelectorSpec,
    },
    /// Restart matching sheep
    Restart {
        /// Which sheep
        selector: SelectorSpec,
    },
    /// Replace each matching sheep with a fresh instance of the same app, one
    /// instance of an app at a time, so the app has a window in which it can
    /// stay reachable across the swap
    Reload {
        /// Which sheep. No default: a reload replaces running processes.
        selector: SelectorSpec,
    },
    /// Stop + deregister matching sheep
    Delete {
        /// Which sheep
        selector: SelectorSpec,
    },
    /// Set how many instances one app runs (see `shep stock`).
    ///
    /// Takes a name where every other verb takes a [`SelectorSpec`]:
    /// `instances` is a per-app number and slots are allocated against the
    /// same-name group, so a selector matching two apps could mean four of
    /// each or four in total.
    ///
    /// The count is absolute: two operators sending `+2` against the same
    /// app would get a number neither asked for.
    Scale {
        /// The app's name, exactly as its config spells it. Not a selector: no
        /// `all`, no regex, no `fold:`.
        name: String,
        /// How many instances the app has when this returns. `0` is refused
        /// with [`RpcErrorCode::InvalidConfig`]: `shep delete` is the verb
        /// for removing an app.
        count: u32,
    },
    /// Attach a short marker to `sheep` for `shep flock` to paint, or clear
    /// it with `None`.
    ///
    /// By name, not a selector: a smit belongs to a sheep, and every
    /// instance of that name shows it, one spawned after the paint included.
    ///
    /// Held in memory and scoped to the connection that sent it, so a
    /// publisher republishes rather than publishing on change.
    SetSmit {
        /// Which sheep.
        sheep: String,
        /// The marker, or `None` to clear it.
        smit: Option<Smit>,
    },
    /// Reopen every matched sheep's log files, for an external rotator that
    /// has renamed them (`create`-mode rotation)
    Reopen {
        /// Which sheep
        selector: SelectorSpec,
    },
    /// Empty every matched sheep's log files: flush what is still pending,
    /// then truncate the recorded paths
    Flush {
        /// Which sheep. No default: this destroys log data.
        selector: SelectorSpec,
    },
    /// Send a named action to every matched sheep over its shepherd channel
    /// and report what each app says back (see `shep trigger`).
    Trigger {
        /// Which sheep. No default, matching every other verb that reaches
        /// a running process.
        selector: SelectorSpec,
        /// The action name. Free-form: the daemon never declares, parses, or
        /// validates it, and an app that does not recognize the name is
        /// expected to say so in its own reply.
        action: String,
        /// Argument text, passed through to the app verbatim. One opaque
        /// string, matching the shepherd channel's own `action` message this
        /// becomes.
        params: Option<String>,
    },
    /// Deliver one signal to every matched sheep's own process, never its
    /// process group (see `shep signal`).
    Signal {
        /// Which sheep. No default, matching every other verb that reaches
        /// a running process.
        selector: SelectorSpec,
        /// The signal's name, as
        /// [`OperatorSignal`](crate::signals::OperatorSignal) spells it. The
        /// `SIG` prefix and the case are both optional; a name outside the
        /// grammar answers [`RpcErrorCode::InvalidConfig`].
        signal: String,
    },
    /// Write one line to every matched sheep's stdin (see `shep whisper`).
    SendLine {
        /// Which sheep. No default, matching every other verb that reaches a
        /// running process.
        selector: SelectorSpec,
        /// The line, without its terminator: the shepherd appends exactly
        /// one `\n` when it writes.
        ///
        /// A line containing an embedded newline is refused
        /// ([`RpcErrorCode::InvalidConfig`]): it would deliver two commands
        /// where the operator typed one.
        line: String,
    },
    /// Write the muster roll now, bypassing the snapshot writer's debounce
    SaveRoll,
    /// Assemble the flock from the muster roll on disk: start every app the
    /// roll recorded running, leaving every app the flock already has exactly
    /// as it stands
    Muster,
    /// Ask for one dog's `[dog.<name>]` section, as the dog itself parses it
    DogConfig {
        /// The dog's name: the config key, not a selector
        name: String,
    },
    /// Start one dog now, marking it as coming from `source`
    EnableDog {
        /// The dog's name
        name: String,
        /// Where its binary comes from
        source: DogSource,
    },
    /// Stop and deregister one dog
    ///
    /// Answers [`Response::Deleted`]: disabling deregisters exactly as
    /// `Delete` does.
    DisableDog {
        /// The dog's name
        name: String,
    },
    /// Ask which dogs this daemon has given up on, and which it is still
    /// waiting to hear from (`shep daemon reload`).
    ///
    /// Read-only, and about this daemon's own handshakes: take the reading
    /// after a reload, not before one. Never sent to an older daemon, on
    /// [`Self::HandoverFitness`]'s terms.
    ///
    /// Answers [`Response::DogStaleness`].
    DogStaleness,
    /// Ask whether this daemon could hand its flock to a successor in place,
    /// rather than stopping it and starting it again (`shep daemon reload`).
    ///
    /// Read-only: the handover itself is triggered by a signal, which reaches
    /// a daemon that refuses the client at the handshake.
    ///
    /// Answers [`Response::HandoverFitness`]. A refusal is a feature the
    /// running daemon cannot carry, not an error: the caller falls back to a
    /// stop-and-start and prints the reason. Never sent to an older daemon:
    /// shep-cli's `commands::daemon` gates it on the crate version the
    /// handshake reported.
    HandoverFitness,
    /// Graceful daemon shutdown
    KillDaemon,
    /// Subscribe this connection to bus topics (glob patterns)
    Subscribe {
        /// Topic globs, e.g. `process.*`
        topics: Vec<String>,
    },
    /// A request kind this build has not been taught.
    ///
    /// `#[serde(other)]`, which serde allows here because `Request` is
    /// internally tagged and this variant carries nothing. The unknown
    /// body's own fields are discarded: the only thing to do with a
    /// request we cannot name is refuse it, and the refusal needs the
    /// envelope's id rather than the body.
    #[serde(other)]
    Unrecognized,
}

/// Where a dog came from: this binary, or one an operator adopted.
///
/// Carried on [`ProcessInfo::dog`], so a listing distinguishes the two
/// populations without a second request.
// wire format: changing existing variants is a breaking change
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum DogSource {
    /// An argv branch of the shep binary itself (`shep dog <name>`).
    BuiltIn,
    /// A binary an operator adopted, run at the daemon's own trust level.
    Adopted {
        /// The binary's path, exactly as the operator gave it to `adopt`.
        path: String,
    },
}

/// One process the OS reports as a descendant of a sheep.
///
/// Not the set of processes that die with the sheep: this is a parent-pid
/// walk, where the stop ladder acts on the process group. A lamb that forks
/// and exits leaves children re-parented to init, out of this list and still
/// in the group; a `setsid()` grandchild stays in the list and leaves the
/// group.
///
/// `name` is the executable's name (`node`, `sh`), never argv, which carries
/// credentials and would ride into `shep describe --format json`. Build one
/// with [`Self::new`].
// wire format: changing this is a breaking change
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Lamb {
    /// The lamb's own pid.
    pub pid: u32,
    /// The executable's name, as the OS reports it. Never its command line.
    pub name: String,
}

impl Lamb {
    /// One lamb.
    #[must_use]
    pub fn new(pid: u32, name: impl Into<String>) -> Self {
        Self {
            pid,
            name: name.into(),
        }
    }
}

/// Why a sheep's process most recently stopped existing under this daemon.
///
/// Behind [`ProcessInfo::last_exit`]'s own `Option`, so `None` there means
/// never exited. Ordinarily exactly one of `code`/`signal` is `Some`,
/// mirroring the OS's `WIFEXITED`/`WIFSIGNALED` split; both `None` together
/// is legal and means this daemon recorded an exit it could not
/// characterize.
// wire format: changing this is a breaking change
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct ExitInfo {
    /// The process's own exit code, set on a normal exit (`WIFEXITED`).
    pub code: Option<i32>,
    /// The raw unix signal number that ended the process, set when it did
    /// not exit on its own (`WIFSIGNALED`). An operator's own `shep stop`
    /// counts: the process still stopped by a signal.
    ///
    /// Platform-specific, and never rendered as a name here; that is an
    /// OS-aware layer's job.
    pub signal: Option<i32>,
}

/// Snapshot of one sheep for listings and events
///
/// Construct one with [`ProcessInfo::builder`]: `#[non_exhaustive]` forbids
/// a struct literal outside this crate, though not inside it. The fields
/// stay `pub`.
// wire format: changing this is a breaking change. No `Eq`: `cpu_percent` is
// an `f32`. Paths travel as `String`, since serde's `PathBuf` refuses a
// non-UTF-8 path and would blank a whole `Reply`. Every added field is an
// `Option`, so a peer built before it sends no key and `None` reads as unknown.
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ProcessInfo {
    /// Stable numeric id
    pub id: u32,
    /// Sheep name
    pub name: String,
    /// Lifecycle status
    pub status: ProcStatus,
    /// OS pid while running
    pub pid: Option<u32>,
    /// Restart count since registration
    pub restarts: u32,
    /// Milliseconds since last successful start
    pub uptime_ms: u64,
    /// Fold membership
    pub fold: Option<String>,
    /// Names this sheep waits for at a staged start, from its
    /// `depends_on`. Empty both when the sheep declares none and when the
    /// peer daemon predates the field.
    #[serde(default)]
    pub depends_on: Vec<String>,
    /// Resolved stdout log path: the app's explicit
    /// [`AppConfig::out_file`] when it set one, else the daemon-derived
    /// default. `None` only when the peer daemon predates this field.
    pub out_file: Option<String>,
    /// Resolved stderr log path, resolved exactly as [`Self::out_file`]
    pub err_file: Option<String>,
    /// Tree CPU as a percentage of one core, over the window since the
    /// daemon's last periodic sample. `None` when the sheep is not running,
    /// when it has been up for less than one sampling window, or when the
    /// peer daemon predates the field; all three render as unknown, never as
    /// zero. A value over 100 is a tree using more than one core.
    pub cpu_percent: Option<f32>,
    /// Tree resident set size in bytes, current as of the reply. `None`
    /// under the same three conditions as [`Self::cpu_percent`], minus the
    /// window one: memory needs no baseline.
    pub memory_bytes: Option<u64>,
    /// Set when this entry is a dog, naming where the dog came from;
    /// `None` for a sheep.
    ///
    /// Two cases, not [`Self::cpu_percent`]'s three: "not a dog" is the true
    /// answer whether the peer predates the field or the entry is a sheep.
    pub dog: Option<DogSource>,
    /// The processes the OS reports as descendants of this sheep, or `None`
    /// when this reply did not walk for them.
    ///
    /// `None` covers two cases: this reply is not a `Describe` (only
    /// `Describe` walks), or the peer daemon predates the field.
    /// `Some(vec![])` is the third: walked, and this sheep has no children.
    ///
    /// Read [`Lamb`]'s own doc before rendering this: the list is not the set
    /// of processes a stop kills, and any output built from it has to say so.
    pub lambs: Option<Vec<Lamb>>,
    /// How this sheep's process most recently stopped existing under this
    /// daemon. `None` while it has never exited under this daemon, and when
    /// the peer daemon predates the field.
    ///
    /// Sticky across a respawn: it answers why the sheep last stopped, not
    /// whether it is stopped now, and updates only on the next exit.
    pub last_exit: Option<ExitInfo>,
    /// The marker a dog has asked to have painted beside this sheep, or
    /// `None` when no dog has painted one, which also covers a peer daemon
    /// that predates the field.
    ///
    /// A `String` rather than a [`Smit`]: this is a report, and the
    /// validation that makes it safe to print happened at the daemon's
    /// ingress. Every instance of a name shows the same marker, since smits
    /// are keyed by sheep name.
    pub smit: Option<String>,
    /// Which instance slot of its app this sheep occupies, counting from 0.
    ///
    /// `None` when the peer daemon predates the field. Not a bare `u32`
    /// defaulted to 0: an app stocked to four instances would report four
    /// rows all claiming slot 0.
    pub instance: Option<u32>,
    /// Whether this dog has completed a handshake with the shepherd that is
    /// reporting it, and not been refused since; `None` for a sheep.
    ///
    /// `None` on [`Self::dog`]'s two-case terms: a sheep has no connection
    /// to the shepherd, so "no handshake fact to report" is the true answer
    /// for a sheep and for a peer that predates the field alike.
    ///
    /// `Some(false)` is the one that matters: a dog on a protocol this
    /// shepherd refuses is alive, which is all [`Self::status`] reports, and
    /// not doing its job. A fact and not a verdict, though: a dog spawned a
    /// moment ago has not handshaken yet and is healthy.
    pub handshook: Option<bool>,
    /// Whether the reporting shepherd has given up on this dog: restarted it
    /// once for never answering, watched that not help, and stopped
    /// restarting it. `None` for a sheep, on [`Self::handshook`]'s terms.
    ///
    /// Not derivable from [`Self::handshook`]. `Some(false)` there covers
    /// both a dog spawned three seconds ago that has not dialled back and one
    /// this shepherd has permanently stopped restarting; the first needs
    /// nothing done and the second is an incident.
    ///
    /// A fact and not a verdict: it says the shepherd stopped, never why.
    /// The why is in that dog's own log (`shep bleats <dog>`).
    pub dog_stale: Option<bool>,
    /// The [`AppConfig`] field names this sheep's spec differs from a load's
    /// parked config for, in field-name order. `None` when nothing is parked,
    /// and when the peer daemon predates the field.
    ///
    /// Names only, never values, as [`SheepDrift::fields`] carries them: a
    /// differing `env` reports `"env"` and stops there. `shep reload`
    /// promotes a parked config.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pending: Option<Vec<String>>,
    /// The [`AppConfig`] field names an operator has set on this sheep that
    /// its current Flockfile does not declare, in field-name order. `None`
    /// when there is nothing to report, and when the peer daemon predates
    /// the field.
    ///
    /// Names only, never values, for [`Self::pending`]'s reason:
    /// [`crate::overrides::AppOverrides::fields`] can hold an `env` value.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub overridden: Option<Vec<String>>,
    /// The sheep's `max_memory` ceiling in bytes, when it has one.
    ///
    /// Additive, like [`Self::instance`] and [`Self::handshook`] before it, so
    /// neither `PROTOCOL_VERSION` nor `SCHEMA_VERSION` moves: an older payload
    /// decodes with it absent and an older client ignores it. Lookout's
    /// `MEM/CEIL` gauge is the only reader; `None` draws an all-tail bar
    /// rather than guessing a denominator.
    pub max_memory: Option<u64>,
}

/// Orders one flock listing the way every operator-facing surface presents
/// one: `(name, instance, id)`.
///
/// Name first: an id is assigned at registration and a `delete all` plus a
/// fresh start renumbers the flock, where a name survives. A name is not a
/// total order on its own, so the id breaks the tie and stays an addressing
/// key (`shep stop 11`).
///
/// A listing whose rows all carry `None` for the slot collapses to
/// `(name, id)`, since `None` sorts before every `Some`.
pub fn sort_flock(listing: &mut [ProcessInfo]) {
    listing.sort_unstable_by(|a, b| {
        (a.name.as_str(), a.instance, a.id).cmp(&(b.name.as_str(), b.instance, b.id))
    });
}

impl ProcessInfo {
    /// Starts a builder for one sheep's row.
    ///
    /// The three arguments are the fields no row can omit and no reader can
    /// default.
    ///
    /// No `#[must_use]`: [`ProcessInfoBuilder`] carries one, which clippy's
    /// `double_must_use` lint treats as covering this return too.
    pub fn builder(id: u32, name: impl Into<String>, status: ProcStatus) -> ProcessInfoBuilder {
        ProcessInfoBuilder {
            info: Self {
                id,
                name: name.into(),
                status,
                pid: None,
                restarts: 0,
                uptime_ms: 0,
                fold: None,
                depends_on: Vec::new(),
                out_file: None,
                err_file: None,
                cpu_percent: None,
                memory_bytes: None,
                dog: None,
                lambs: None,
                last_exit: None,
                smit: None,
                instance: None,
                handshook: None,
                dog_stale: None,
                pending: None,
                overridden: None,
                max_memory: None,
            },
        }
    }
}

/// Builds a [`ProcessInfo`], which is `#[non_exhaustive]` and so cannot be
/// written as a struct literal outside this crate.
///
/// Every setter takes the field's own type, `Option` included, so a caller
/// already holding `Option<u32>` writes `.pid(entry.pid())` rather than an
/// `if let` ladder. A setter is skipped, not passed `None`, when a row has
/// nothing to say about that field; the skipped defaults are the ones a
/// not-yet-running sheep has.
#[derive(Debug, Clone)]
#[must_use = "a builder that is never `build`-ed produces no ProcessInfo"]
pub struct ProcessInfoBuilder {
    info: ProcessInfo,
}

impl ProcessInfoBuilder {
    /// Sets the OS pid; `None` while the sheep is not running.
    pub fn pid(mut self, pid: Option<u32>) -> Self {
        self.info.pid = pid;
        self
    }

    /// Sets the restart count since registration.
    pub fn restarts(mut self, restarts: u32) -> Self {
        self.info.restarts = restarts;
        self
    }

    /// Sets milliseconds since the last successful start.
    pub fn uptime_ms(mut self, uptime_ms: u64) -> Self {
        self.info.uptime_ms = uptime_ms;
        self
    }

    /// Sets fold membership.
    pub fn fold(mut self, fold: Option<String>) -> Self {
        self.info.fold = fold;
        self
    }

    /// Sets the names this sheep waits for at a staged start.
    pub fn depends_on(mut self, depends_on: Vec<String>) -> Self {
        self.info.depends_on = depends_on;
        self
    }

    /// Sets the resolved stdout log path.
    pub fn out_file(mut self, out_file: Option<String>) -> Self {
        self.info.out_file = out_file;
        self
    }

    /// Sets the resolved stderr log path.
    pub fn err_file(mut self, err_file: Option<String>) -> Self {
        self.info.err_file = err_file;
        self
    }

    /// Sets tree CPU as a percentage of one core.
    pub fn cpu_percent(mut self, cpu_percent: Option<f32>) -> Self {
        self.info.cpu_percent = cpu_percent;
        self
    }

    /// Sets tree resident set size in bytes.
    pub fn memory_bytes(mut self, memory_bytes: Option<u64>) -> Self {
        self.info.memory_bytes = memory_bytes;
        self
    }

    /// Marks this row a dog and names where the dog came from.
    pub fn dog(mut self, dog: Option<DogSource>) -> Self {
        self.info.dog = dog;
        self
    }

    /// Sets the sheep's lamb list; `None` when this reply did not walk for one.
    pub fn lambs(mut self, lambs: Option<Vec<Lamb>>) -> Self {
        self.info.lambs = lambs;
        self
    }

    /// Sets how this sheep's process most recently stopped; `None` while it
    /// has never exited under this daemon.
    pub fn last_exit(mut self, last_exit: Option<ExitInfo>) -> Self {
        self.info.last_exit = last_exit;
        self
    }

    /// Sets the marker a dog has painted on this sheep; `None` when none has.
    pub fn smit(mut self, smit: Option<String>) -> Self {
        self.info.smit = smit;
        self
    }

    /// Sets the instance slot; `None` when the peer daemon predates the field.
    pub fn instance(mut self, instance: Option<u32>) -> Self {
        self.info.instance = instance;
        self
    }

    /// Sets whether this dog has handshaken with the shepherd; `None` for a
    /// sheep, which has no handshake to report.
    pub fn handshook(mut self, handshook: Option<bool>) -> Self {
        self.info.handshook = handshook;
        self
    }

    /// Sets whether the shepherd has given up restarting this dog; `None`
    /// for a sheep, which is never given up on.
    pub fn dog_stale(mut self, dog_stale: Option<bool>) -> Self {
        self.info.dog_stale = dog_stale;
        self
    }

    /// Sets the field names a load has parked for this sheep's next spawn;
    /// `None` when nothing is parked.
    pub fn pending(mut self, pending: Option<Vec<String>>) -> Self {
        self.info.pending = pending;
        self
    }

    /// Sets the field names an operator has overridden on this sheep;
    /// `None` when there is nothing to report.
    pub fn overridden(mut self, overridden: Option<Vec<String>>) -> Self {
        self.info.overridden = overridden;
        self
    }

    /// Sets the sheep's `max_memory` ceiling in bytes; `None` when it has no
    /// ceiling configured.
    pub fn max_memory(mut self, max_memory: Option<u64>) -> Self {
        self.info.max_memory = max_memory;
        self
    }

    /// Finishes the row.
    #[must_use]
    pub fn build(self) -> ProcessInfo {
        self.info
    }
}

/// What happened when the daemon tried to deliver one sheep's triggered
/// action.
// wire format: changing existing variants is a breaking change
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum ActionOutcome {
    /// The app answered on the shepherd channel.
    Replied {
        /// The reply body, exactly as the app sent it.
        body: String,
    },
    /// The sheep had no reachable shepherd channel for the daemon to
    /// deliver the action over.
    NoChannel,
    /// The sheep is a reload drainee, mid-swap and on its way out, so the
    /// daemon skipped it rather than deliver the action to a process already
    /// being replaced.
    Skipped,
    /// The daemon delivered the action, but no reply arrived before the
    /// app's configured action timeout elapsed.
    TimedOut,
}

/// One matched sheep's row in a `Trigger` reply.
///
/// Not a [`ProcessInfo`]: a reply body has nowhere to live on one.
/// [`Self::outcome`] is per-row, since the selector grammar makes a mixed
/// flock the normal case.
// wire format: changing this is a breaking change
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ActionReply {
    /// The sheep's stable id.
    pub id: u32,
    /// The sheep's name.
    pub name: String,
    /// What happened when the daemon tried to deliver the action.
    pub outcome: ActionOutcome,
}

/// What happened when the shepherd tried to deliver one signal.
// wire format: changing existing variants is a breaking change
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum SignalOutcome {
    /// The kernel accepted the signal for this sheep's pid.
    ///
    /// Says the signal was delivered, not that the app did anything with it.
    /// A signal the app blocks or ignores is `Delivered` too.
    Delivered,
    /// The sheep is registered but has no live process to signal: stopped,
    /// errored, or waiting out a restart backoff.
    NotRunning,
    /// The kernel refused the delivery; carries its reason (`ESRCH` for a
    /// process reaped between the lookup and the syscall, `EPERM` for one this
    /// daemon may not signal).
    Failed {
        /// The refusal, as the OS worded it.
        reason: String,
    },
}

/// One matched sheep's row in a `Signal` reply.
///
/// Per-row like [`ActionReply`]: the selector grammar makes a mixed flock
/// the normal case.
// wire format: changing this is a breaking change
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SignalReply {
    /// The sheep's stable id.
    pub id: u32,
    /// The sheep's name.
    pub name: String,
    /// What happened when the shepherd tried to deliver the signal.
    pub outcome: SignalOutcome,
}

/// What happened when the shepherd tried to write one line to a sheep's stdin.
// wire format: changing existing variants is a breaking change
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum LineOutcome {
    /// The line was written to the pipe and flushed.
    ///
    /// Says the bytes left the shepherd, not that the app read them. A pipe
    /// holds 64 KiB before it blocks.
    Sent,
    /// The sheep has no stdin pipe: its config does not set `stdin = true`, or
    /// it is not running.
    ///
    /// One outcome for two causes: both answer "there is no pipe here".
    NoStdin,
    /// The shepherd had a pipe and did not confirm a write to it; carries
    /// why.
    ///
    /// Three shapes reach it: the write failed (the far end is gone), the
    /// line found the sheep's queue already full, or the write did not
    /// finish inside the shepherd's own bound. The reason names which.
    ///
    /// A timed-out write is not a promise the line was never written: the
    /// bytes may be part-written into a pipe the app is not draining, and
    /// land in full the moment it drains. A line still queued behind that one
    /// is dropped once its caller gives up, so treat a retry as a second
    /// command.
    NotWritten {
        /// What went wrong, in plain English.
        reason: String,
    },
}

/// One matched sheep's row in a `SendLine` reply.
///
/// Per-row like [`ActionReply`] and [`SignalReply`].
// wire format: changing this is a breaking change
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LineReply {
    /// The sheep's stable id.
    pub id: u32,
    /// The sheep's name.
    pub name: String,
    /// What happened.
    pub outcome: LineOutcome,
}

/// A dog's `[dog.<name>]` config section, carried as TOML text.
///
/// Travels over the socket rather than the child's environment: a dog's
/// section routinely holds webhook credentials, and the socket keeps them
/// out of the process table and out of crash dumps. The manual `Debug`
/// below prints only a length, since [`Response`] derives `Debug`.
///
/// [`Self::as_str`] is the only way out: a `Deref<Target = str>` would hand
/// the type `ToString` and defeat that `Debug`.
///
/// `#[serde(transparent)]`: the wire representation is a bare `String`.
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct DogSectionToml(String);

impl DogSectionToml {
    /// The TOML text, empty when the file has no such section.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl From<String> for DogSectionToml {
    fn from(toml: String) -> Self {
        Self(toml)
    }
}

/// Prints a length, never the section body. Pinned as an exact string by
/// `dog_section_toml_debug_does_not_leak`.
impl fmt::Debug for DogSectionToml {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "DogSectionToml(<{} bytes>)", self.0.len())
    }
}

/// One environment variable's value, on its way to a sheep.
///
/// A newtype for one reason, the same one [`DogSectionToml`] exists for: an
/// env value is the single most secret-dense thing a client can send this
/// daemon (a database URL, an API token, a signing key), and a derived
/// `Debug` on [`Request`] would print it in the clear the moment anything
/// logs a request. Every other secret-bearing field on this wire is already
/// protected by its inner type ([`AppConfig`]'s own manual `Debug` prints
/// `env: <N vars>`), and a bare `String` here would have been the first
/// field in the enum without that protection.
///
/// One direction only. Nothing ever sends one back: [`Request::SheepConfig`]
/// answers with the env keys and no values at all.
///
/// [`Self::as_str`] is the only way out, for the reason
/// [`DogSectionToml`] gives: a `Deref<Target = str>` would hand the type
/// `ToString` too, and `.to_string()` would return the value in the clear,
/// defeating the redacted `Debug` below.
///
/// `#[serde(transparent)]` makes the wire representation identical to a
/// bare `String`, so this newtype changes nothing about
/// [`crate::protocol::PROTOCOL_VERSION`] or the pinned snapshot fixtures.
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct EnvValue(String);

impl EnvValue {
    /// The value.
    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl From<String> for EnvValue {
    fn from(value: String) -> Self {
        Self(value)
    }
}

/// Debug prints a length and never the value (IR-41); see the type doc for
/// why. Exact-string-tested below (`env_value_debug_does_not_leak`) so a
/// future `#[derive(Debug)]` fails that test instead of silently reopening
/// the leak.
impl fmt::Debug for EnvValue {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "EnvValue(<{} bytes>)", self.0.len())
    }
}

/// One registered sheep whose stored config differs from a caller's copy:
/// the answer [`Request::ConfigDrift`] is asking for
///
/// Field names only, never their values. This is printed at an operator,
/// and [`AppConfig::env`](crate::config::AppConfig::env) carries secrets,
/// so a differing `env` reports `"env"` and nothing more. `Debug` is
/// derived: there is nothing here to redact.
// wire format: changing field names is a breaking change
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SheepDrift {
    /// The sheep's name. Both configs share it by construction: it is what
    /// matched them to each other.
    pub name: String,
    /// The [`AppConfig`] fields that differ, in field-name order. Never
    /// empty: a sheep with nothing to report is left out of the answer.
    pub fields: Vec<String>,
}

impl SheepDrift {
    /// Builds one sheep's report.
    #[must_use]
    pub fn new(name: impl Into<String>, fields: Vec<String>) -> Self {
        Self {
            name: name.into(),
            fields,
        }
    }
}

/// What one app's [`Request::ApplyConfig`] did: the answer a load owes the
/// operator who ran it
///
/// One of these per app the request named, found or not and changed or not.
///
/// [`Self::applied`] and [`Self::pending`] carry field names only, never
/// their values, as [`SheepDrift`] does; the merged config never reaches a
/// client. [`Self::refused`] is prose and is scoped out of that rule: it
/// quotes values out of the file the caller just sent, never out of the
/// flock's stored config. `Debug` is derived on that basis: nothing here
/// needs redacting.
// wire format: changing field names is a breaking change
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SheepApplied {
    /// The sheep's name, exactly as the request spelled it.
    pub name: String,
    /// Fields now in force, in field-name order. Empty when the load changed
    /// nothing the daemon could act on immediately.
    pub applied: Vec<String>,
    /// Fields the app picks up at its next spawn, in field-name order. Empty
    /// when nothing is waiting.
    ///
    /// `shep reload <name>` promotes them; a client rendering this list says
    /// so, since a pending list with no remedy beside it cannot be acted on.
    pub pending: Vec<String>,
    /// Why some or all of this app's change did not land, in the daemon's own
    /// words, or `None` when the whole of it did.
    ///
    /// Not the same question as the two lists being empty: a refusal raised
    /// before anything was touched leaves both empty, and so does a load with
    /// nothing to do. It is a sentence rather than a code because the message
    /// is what tells them apart.
    pub refused: Option<String>,
}

impl SheepApplied {
    /// Builds one app's report.
    #[must_use]
    pub fn new(
        name: impl Into<String>,
        applied: Vec<String>,
        pending: Vec<String>,
        refused: Option<String>,
    ) -> Self {
        Self {
            name: name.into(),
            applied,
            pending,
            refused,
        }
    }
}

/// One app a multi-sheep reload or restart could not accept, and why
///
/// A staged walk asks the supervisor per app, so an app already reloading is
/// refused on its own while the rest of the fold goes ahead, and so is one
/// that left the flock after the walk was planned. One of these per refused
/// app rides back in [`Response::Reloading`] or [`Response::Restarted`],
/// which is what lets the client name the app and exit non-zero instead of
/// printing a table with a row quietly missing from it.
///
/// [`Self::reason`] is the daemon's own sentence rather than a code, the
/// rule [`SheepApplied::refused`] takes and for its reason: the class of
/// refusal is not on the wire, and the message is what tells two of them
/// apart. `Debug` is derived; a name and a refusal sentence carry no env,
/// no path and no argument vector.
// wire format: changing field names is a breaking change
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SheepRefusal {
    /// The app's name, as the walk that planned the reload or the restart
    /// spelled it.
    pub name: String,
    /// Why that app was refused, in the daemon's own words.
    pub reason: String,
}

impl SheepRefusal {
    /// Builds one app's refusal.
    #[must_use]
    pub fn new(name: impl Into<String>, reason: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            reason: reason.into(),
        }
    }
}

/// One sheep's effective config as a pane sees it: every field but env's
/// values, plus which fields an operator has overridden and which are
/// waiting on a respawn.
///
/// The answer to [`Request::SheepConfig`], and the one reply in this module
/// that carries a whole [`AppConfig`]. [`SheepApplied`] deliberately carries
/// field names alone, and the difference is what each is for: that one is
/// printed at an operator who already has the file, this one feeds a pane
/// that is about to edit fields it has to be able to show first.
// wire format: changing field names is a breaking change
//
// `#[non_exhaustive]`: shep-core is a published library and a sixth field
// would otherwise break an out-of-tree consumer's construction of this with
// no version bump to say so (IR-20). [`SheepConfigView::new`] is how the
// daemon builds one, and it is what enforces the emptied `env`.
#[non_exhaustive]
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SheepConfigView {
    /// The sheep's name.
    pub name: String,
    /// The effective config with `env` cleared. Every remaining field is
    /// operator-supplied policy the pane is about to let them edit, so
    /// withholding a value would make the pane unusable while protecting
    /// nothing.
    pub config: AppConfig,
    /// The env keys, so the pane can list them. Never the values.
    pub env_keys: Vec<String>,
    /// Field names an operator has set that the Flockfile does not declare.
    pub overridden: Vec<String>,
    /// Field names parked until the next respawn.
    pub pending: Vec<String>,
}

impl SheepConfigView {
    /// Builds one, clearing `env` and recording its keys.
    ///
    /// The clearing happens here rather than at the one call site, so a
    /// second caller cannot forget it: this constructor is the only way to
    /// build the type outside this crate, since `#[non_exhaustive]` blocks
    /// a literal.
    #[must_use]
    pub fn new(mut config: AppConfig, overridden: Vec<String>, pending: Vec<String>) -> Self {
        let env_keys = config.env.keys().cloned().collect();
        config.env.clear();
        Self {
            name: config.name.clone(),
            config,
            env_keys,
            overridden,
            pending,
        }
    }
}

/// Redacted (IR-41): `config` carries `args` and `cwd`, which routinely hold
/// a token or a home directory, and this type is what a `{:?}` on a
/// [`Response`] would print. The three lists are counted rather than named
/// for the same reason: `env_keys` is a key set, which is itself worth
/// keeping out of a log.
impl fmt::Debug for SheepConfigView {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
            "SheepConfigView {{ name: {:?}, env_keys: {}, overridden: {}, pending: {} }}",
            self.name,
            self.env_keys.len(),
            self.overridden.len(),
            self.pending.len()
        )
    }
}

/// One RPC response (pairs with [`Request`] variants)
///
/// Ten variants carry a bare `Vec<ProcessInfo>`. Do not collapse them into
/// one: each names which request it answers, which is what lets a variant
/// diverge without a protocol bump. `Reloading` already means an acceptance
/// rather than a result, `Scaled` only the survivors of a scale-down, and
/// `Mustered` every sheep of every restored app rather than what this call
/// started.
// wire format: changing existing variants is a breaking change.
// `large_enum_variant` allowed, not fixed: clippy's remedy is to box
// `DogStarted`'s payload, a source break for every
// `Response::DogStarted(info)` in and out of this workspace.
#[allow(clippy::large_enum_variant)]
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", content = "data", rename_all = "snake_case")]
#[non_exhaustive]
pub enum Response {
    /// Answer to `Ping`
    Pong,
    /// Answer to `ListFlock`
    Flock(Vec<ProcessInfo>),
    /// Answer to `Describe`
    Described(Vec<ProcessInfo>),
    /// Answer to `Start`
    Started(Vec<ProcessInfo>),
    /// Answer to `Add`: one row per app the request named, registered and
    /// spawning nothing.
    ///
    /// A row here can still be `Online`: `Add` is idempotent by name, so the
    /// reply describes the membership the request leaves behind.
    Added(Vec<ProcessInfo>),
    /// Answer to `ConfigDrift`: one entry per app that is registered under a
    /// config different from the one asked about, and no entry for anything
    /// else. An empty vector means every app asked about either matches or
    /// is not registered at all.
    Drifted(Vec<SheepDrift>),
    /// Answer to `ApplyConfig`: one entry per app the request named, in the
    /// order it named them, the refused and the unchanged included.
    ///
    /// Complete where [`Self::Drifted`] is filtered: an app missing from
    /// "what did you do to each of these" looks like one the daemon dropped.
    Applied(Vec<SheepApplied>),
    /// Answer to `SheepConfig`: one sheep's config with `env` emptied and
    /// its keys listed beside it.
    ///
    /// Boxed, and the only variant here that is. This one carries a whole
    /// [`AppConfig`], which is several times the size of anything else in
    /// the enum, and a `Response` is inside a `Reply` which is inside a
    /// [`ServerFrame`](crate::protocol::ServerFrame): without the box,
    /// every frame the daemon sends costs the largest config's worth of
    /// stack for a variant almost none of them use.
    ///
    /// The enum-level `#[allow(clippy::large_enum_variant)]` below does not
    /// cover it, and the difference is the point of that allow's own
    /// argument: boxing `DogStarted` would be a source break for every
    /// `Response::DogStarted(info)` in and out of this workspace, where
    /// this variant has never shipped and so breaks nobody.
    ///
    /// `Box<T>` serializes exactly as `T`, so the wire bytes and the pinned
    /// fixtures are untouched.
    SheepConfig(Box<SheepConfigView>),
    /// Answer to `SetSheepEnv`: the key that was set or removed.
    ///
    /// Never the value, and never the resulting env map. This reply exists
    /// to confirm which key moved, and echoing what was just written back
    /// down a socket would undo the whole point of `SheepConfig` withholding
    /// it (IR-41).
    SheepEnvSet {
        /// The sheep.
        name: String,
        /// The key.
        key: String,
    },
    /// What a [`Request::SetSheepEnvBatch`] did, or would have done.
    ///
    /// Key names only. `set` is what was written, `unchanged` what already
    /// held the same value, `collisions` what held a different one. A
    /// forced request reports a collision in both `set` and `collisions`;
    /// an unforced one that collides reports an empty `set` and wrote
    /// nothing.
    SheepEnvBatch {
        /// The sheep's name.
        name: String,
        /// Keys written.
        set: Vec<String>,
        /// Keys that already held this value.
        unchanged: Vec<String>,
        /// Keys that held a different value.
        collisions: Vec<String>,
    },
    /// Answer to `SetSheepField`: which field moved, and whether the
    /// running child has it.
    ///
    /// Not [`Self::Applied`]'s three lists; the difference is the
    /// request's own shape. `applied`, `pending` and `refused` exist
    /// because `ApplyConfig` carries N apps of M fields, so a caller cannot
    /// otherwise tell which field went where or that one app of eleven was
    /// refused. This request carries one field of one sheep, so `refused`
    /// would be a second way to say no beside the `Err` arm (a client
    /// checking only the `Err` would silently swallow the other), and the
    /// two lists collapse to the one bit that is left.
    ///
    /// That bit is not redundant with the field's own
    /// [`ApplyGroup`](crate::config::ApplyGroup), which the caller already
    /// knows. It is the daemon's answer about state a caller cannot see:
    /// `autostart` is `NextSpawn` and yet reports as in force, because it
    /// is read at muster rather than at a spawn, and a `Live` field whose
    /// config subset will not normalize on its own parks instead of
    /// applying.
    SheepFieldSet {
        /// The sheep.
        name: String,
        /// The field that moved.
        key: String,
        /// `true` when the running child does not have the value yet and
        /// `shep reload <name>` is what promotes it. A client rendering
        /// this says so, the same rule [`SheepApplied::pending`] carries.
        pending: bool,
    },
    /// Answer to `SetDogConfig`: the section was written and the topic
    /// published.
    DogConfigSet {
        /// The dog.
        name: String,
    },
    /// Answer to [`Request::PutSecrets`]: how many entries were stored.
    SecretsPut {
        /// Entry count, after the namespace and environment were replaced.
        accepted: u32,
    },
    /// Answer to `Stop`
    Stopped(Vec<ProcessInfo>),
    /// Answer to `Restart`: the sheep that were restarted, one row each.
    ///
    /// **When the reply arrives depends on how many sheep matched.** One
    /// sheep is answered as soon as its respawn is issued. Two or more are
    /// restarted in dependency order, and the daemon holds each stage until
    /// the apps a later stage waits on are back, so the reply arrives no
    /// sooner than the last stage's respawns and the rows are stitched from
    /// one answer per stage. A client asking for a budget sizes it for the
    /// whole walk, not for one respawn.
    Restarted {
        /// The sheep the restart reached, one row each.
        ///
        /// A row is not a promise the process is up. A respawn that could
        /// not exec is an `errored` row here rather than an entry in
        /// `refused` below: the sheep was reached and the restart was not
        /// refused, it is the child that failed.
        accepted: Vec<ProcessInfo>,
        /// The apps the walk could not restart, empty when it restarted
        /// every one it named.
        ///
        /// Only a walk fills this. A selector matching one app is refused
        /// whole, as the `Err` arm, so a client reading a single-target
        /// restart never sees a row here.
        refused: Vec<SheepRefusal>,
    },
    /// Answer to `Reload`: acceptances, not results.
    ///
    /// One instance costs a readiness wait plus a drain, so a clustered app
    /// outlasts any deadline a client may ask for. Every row is therefore the
    /// sheep as it stood when its own reload was accepted, and the swaps
    /// report themselves on the bus (`process.reload`, `process.reloaded`,
    /// `process.reload_abandoned`). A matched sheep with nothing to replace
    /// is listed as the no-op success it is.
    ///
    /// **When the reply arrives depends on how many sheep matched.** One
    /// sheep is answered as soon as its reload is accepted. Two or more are
    /// reloaded in dependency order, and the daemon holds each stage until
    /// the swaps of the apps a later stage waits on have landed, so the
    /// reply arrives no sooner than the last stage's acceptance and the rows
    /// are stitched from one acceptance per stage. A client asking for a
    /// budget sizes it for the whole walk, not for one acceptance.
    Reloading {
        /// The sheep whose reloads were accepted, one row each.
        accepted: Vec<ProcessInfo>,
        /// The apps the walk could not reload, empty when it reloaded every
        /// one it named.
        ///
        /// Only a walk fills this. A selector matching one app is refused
        /// whole, as the `Err` arm, so a client reading a single-target
        /// reload never sees a row here.
        refused: Vec<SheepRefusal>,
    },
    /// Answer to `Scale`: the app's instances that will remain, one row each,
    /// ordered by [`sort_flock`]. Every row shares one name, so that is slot
    /// order with the id breaking a tie.
    ///
    /// Scaling down, the departing instances are absent even though their
    /// kill ladders are still running; they report themselves on the bus as
    /// `process.delete`.
    Scaled(Vec<ProcessInfo>),
    /// Answer to `SetSmit`: every instance of the named sheep, one row each,
    /// carrying the smit as it now stands.
    SmitPainted(Vec<ProcessInfo>),
    /// Answer to `Delete`: ids removed
    Deleted(Vec<u32>),
    /// Answer to `Reopen`: every matched sheep, running or not. A sheep with
    /// no live log pump has nothing to reopen and is reported as a success,
    /// so this carries the same matches `Describe` would.
    Reopened(Vec<ProcessInfo>),
    /// Answer to `Flush`: one row per matched sheep, running or not, exactly
    /// as [`Self::Reopened`].
    ///
    /// One row per sheep, not per file emptied: several sheep can share one
    /// log path, and the daemon truncates each distinct path once.
    Flushed(Vec<ProcessInfo>),
    /// Answer to `Trigger`: one [`ActionReply`] row per matched sheep, rather
    /// than a flock listing, since `ProcessInfo` has nowhere to hold a reply
    /// body.
    Triggered(Vec<ActionReply>),
    /// Answer to `Signal`: one [`SignalReply`] row per matched sheep.
    ///
    /// Not a flock listing: [`ProcessInfo`] has nowhere to hold a per-sheep
    /// outcome.
    Signalled(Vec<SignalReply>),
    /// Answer to `SendLine`: one [`LineReply`] row per matched sheep.
    SentLine(Vec<LineReply>),
    /// Answer to `SaveRoll`
    RollSaved {
        /// Absolute path of the roll the daemon wrote
        path: String,
        /// How many apps that roll records
        apps: u32,
    },
    /// Answer to `Muster`: every sheep of every app the roll restored, not
    /// only the ones this call spawned.
    ///
    /// Assembling a flock that is already assembled starts nothing, so a
    /// listing of what this call spawned would be indistinguishable from an
    /// empty roll.
    Mustered(Vec<ProcessInfo>),
    /// Answer to `DogConfig`: the dog's own section, rendered back to TOML.
    ///
    /// `toml` is [`DogSectionToml`], whose manual `Debug` keeps the webhook
    /// credentials this text carries out of a `{:?}`-formatted `Response`.
    DogSection {
        /// The `[dog.<name>]` table as TOML text, empty when the file has
        /// no such section
        toml: DogSectionToml,
    },
    /// Answer to `EnableDog`: the dog as it stands now
    DogStarted(ProcessInfo),
    /// Answer to `DogStaleness`: this daemon's own handshake record, split
    /// into the dogs it has given up on and the dogs it is still waiting on.
    ///
    /// Two lists because they answer two questions. `stale` is a finding;
    /// `pending` is a reason to ask again, since a reading taken now would
    /// be a guess about them.
    ///
    /// Names only: two builds differing only in the protocol they speak
    /// report the same crate version.
    DogStaleness {
        /// Dogs this daemon has refused twice: once on the handshake that
        /// bought them a restart from disk, and again after it. It will not
        /// restart them a third time.
        stale: Vec<String>,
        /// Dogs this daemon is still waiting to hear a final answer from: one
        /// whose restart is in flight, or one it supervises that has not
        /// handshook yet. Neither stale nor known healthy.
        pending: Vec<String>,
    },
    /// Answer to `HandoverFitness`: `None` when the whole flock can be
    /// carried across a daemon handover, and otherwise the sentence saying
    /// which sheep cannot be and why.
    ///
    /// A rendered sentence rather than a structured reason: the set of things
    /// a handover cannot carry keeps changing, and the client only prints it.
    HandoverFitness {
        /// Why the flock cannot be handed over in place, or `None` when it
        /// can.
        refusal: Option<String>,
    },
    /// Answer to `Subscribe`
    Subscribed,
    /// Answer to `KillDaemon`
    ShuttingDown,
}

/// A request frame
// wire format: changing this is a breaking change
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Envelope {
    /// Per-connection request id
    pub id: u64,
    /// Client-imposed deadline (daemon aborts work past it)
    pub deadline_ms: Option<u64>,
    /// The request
    pub body: Request,
}

/// A reply frame
///
/// `result` uses serde's stock `Result` representation: the wire carries
/// `{"Ok": ...}` / `{"Err": ...}`, with capitalized keys, pinned by snapshot.
// wire format: changing this is a breaking change
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Reply {
    /// Echoes [`Envelope::id`]
    pub id: u64,
    /// The outcome
    pub result: Result<Response, RpcError>,
}

/// Handshake outcome: `HelloAck` or a typed refusal, since version skew is
/// an error rather than silence. Same `Ok`/`Err` wire shape as
/// [`Reply::result`]; refusals use [`RpcErrorCode::ProtocolMismatch`].
pub type HelloReply = Result<HelloAck, RpcError>;

/// Structured RPC failure
// wire format: changing this is a breaking change
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RpcError {
    /// Machine-readable code
    pub code: RpcErrorCode,
    /// Human-readable message (plain English, no theme)
    pub message: String,
    /// The daemon's own crate version, when it chose to name it.
    ///
    /// Set on a [`RpcErrorCode::ProtocolMismatch`] refusal, the only place a
    /// client can learn it, since [`HelloAck::daemon_version`] never arrives
    /// there. `None` on every other error, and on a refusal from a daemon
    /// built before the field existed, so a reader treats `None` as unknown
    /// and takes the conservative path.
    ///
    /// Absent on the wire rather than `null`, so
    /// [`crate::protocol::PROTOCOL_VERSION`] does not move for it.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub daemon_version: Option<String>,
}

/// Machine-readable RPC error codes
// wire format: changing existing variants is a breaking change
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum RpcErrorCode {
    /// Selector matched nothing
    NotFound,
    /// Config failed validation daemon-side
    InvalidConfig,
    /// Spawn failed (exec error, permissions)
    SpawnFailed,
    /// Handshake protocol version mismatch
    ProtocolMismatch,
    /// Unexpected daemon-side failure
    Internal,
    /// The request's deadline expired before the daemon finished it
    DeadlineExceeded,
    /// The peer asked for something this build does not implement.
    ///
    /// Distinct from `NotFound`, which means a selector matched nothing.
    /// This means the verb itself is unknown here, and the remedy is a
    /// newer shepherd rather than a different selector.
    Unsupported,
    /// A code this build has not been taught.
    ///
    /// Only ever produced by decoding: an unrecognized string falls through
    /// to this variant via `#[serde(other)]` instead of failing the whole
    /// frame. Nothing constructs one to send, which is a call-site
    /// invariant rather than a type-level one: `#[serde(other)]` governs
    /// decoding only, so serializing this would emit `"unrecognized"`.
    #[serde(other)]
    Unrecognized,
}

impl RpcErrorCode {
    /// Every variant, for code that needs to iterate them all.
    ///
    /// `#[non_exhaustive]` forces a `_` arm on any match written outside this
    /// crate, which would swallow a variant added here and never updated
    /// there (`crates/shep-cli/src/exit.rs` maps every code to an exit
    /// status).
    pub const ALL: [Self; 7] = [
        Self::NotFound,
        Self::InvalidConfig,
        Self::SpawnFailed,
        Self::ProtocolMismatch,
        Self::Internal,
        Self::DeadlineExceeded,
        Self::Unsupported,
    ];

    /// Never called; exists so this crate fails to build if a variant is
    /// added to [`RpcErrorCode`] without also being added to [`Self::ALL`].
    ///
    /// A match here is still checked for exhaustiveness, and each arm indexes
    /// a fixed literal position into [`Self::ALL`], so growing the enum
    /// without growing the array is an out-of-bounds constant index.
    #[allow(dead_code)]
    const fn assert_all_lists_every_variant(code: Self) -> Self {
        match code {
            Self::NotFound => Self::ALL[0],
            Self::InvalidConfig => Self::ALL[1],
            Self::SpawnFailed => Self::ALL[2],
            Self::ProtocolMismatch => Self::ALL[3],
            Self::Internal => Self::ALL[4],
            Self::DeadlineExceeded => Self::ALL[5],
            Self::Unsupported => Self::ALL[6],
            // `Unrecognized` is decode-only and never appears in `ALL`;
            // treat it as `Internal` would be treated.
            Self::Unrecognized => Self::ALL[4],
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::AppConfig;
    use crate::protocol::PROTOCOL_VERSION;
    use crate::status::ProcStatus;

    /// A code this build has never heard of must decode, not fail. Without
    /// this, adding any error code is a breaking change for every peer.
    #[test]
    fn an_unknown_error_code_decodes_as_unrecognized() {
        assert_eq!(
            serde_json::from_str::<RpcErrorCode>(r#""invented_next_year""#).unwrap(),
            RpcErrorCode::Unrecognized
        );
    }

    #[test]
    fn every_known_error_code_still_round_trips() {
        for code in [
            RpcErrorCode::NotFound,
            RpcErrorCode::InvalidConfig,
            RpcErrorCode::SpawnFailed,
            RpcErrorCode::ProtocolMismatch,
            RpcErrorCode::Internal,
            RpcErrorCode::DeadlineExceeded,
            RpcErrorCode::Unsupported,
        ] {
            let json = serde_json::to_string(&code).unwrap();
            assert_eq!(serde_json::from_str::<RpcErrorCode>(&json).unwrap(), code);
        }
    }

    /// The fallback absorbs an unknown STRING, not an unknown TYPE. A number
    /// where a code belongs is still a defect worth reporting.
    #[test]
    fn a_non_string_error_code_is_still_an_error() {
        assert!(serde_json::from_str::<RpcErrorCode>("42").is_err());
    }

    /// The id has to survive a body this build cannot name, or the daemon
    /// has nothing to address a refusal to.
    #[test]
    fn an_unknown_request_kind_keeps_the_envelope_id() {
        let envelope: Envelope = serde_json::from_str(
            r#"{"id":42,"deadline_ms":null,"body":{"kind":"from_the_future","extra":{"a":1}}}"#,
        )
        .unwrap();
        assert_eq!(envelope.id, 42);
        assert_eq!(envelope.body, Request::Unrecognized);
    }

    fn sample_info() -> ProcessInfo {
        ProcessInfo {
            id: 3,
            name: "web".to_string(),
            status: ProcStatus::Online,
            pid: Some(4242),
            restarts: 1,
            uptime_ms: 60_000,
            fold: Some("backend".to_string()),
            // Left empty: this fixture feeds `reply_wire_snapshots` and
            // `bus_event_wire_snapshots`, so a non-empty value moves pinned
            // bytes.
            depends_on: Vec::new(),
            out_file: Some("/home/ada/.shep/logs/web-0-out.log".to_string()),
            err_file: Some("/home/ada/.shep/logs/web-0-err.log".to_string()),
            // 12.5: an insta JSON snapshot is stable across platforms only
            // for a float the binary representation holds exactly.
            cpu_percent: Some(12.5),
            memory_bytes: Some(48 * 1024 * 1024),
            dog: None,
            lambs: None,
            last_exit: Some(ExitInfo {
                code: Some(1),
                signal: None,
            }),
            smit: None,
            instance: None,
            handshook: None,
            dog_stale: None,
            // Left at the builder's default: this fixture feeds
            // `reply_wire_snapshots` and `bus_event_wire_snapshots`, so a
            // `Some(..)` moves pinned bytes.
            pending: None,
            overridden: None,
            max_memory: Some(512 * 1024 * 1024),
        }
    }

    #[test]
    fn a_builder_with_nothing_set_is_a_sheep_that_has_not_run() {
        let info = ProcessInfo::builder(3, "web", ProcStatus::Stopped).build();

        assert_eq!(info.id, 3);
        assert_eq!(info.name, "web");
        assert_eq!(info.status, ProcStatus::Stopped);
        assert_eq!(info.pid, None);
        assert_eq!(info.restarts, 0);
        assert_eq!(info.uptime_ms, 0);
        assert_eq!(info.fold, None);
        assert_eq!(info.out_file, None);
        assert_eq!(info.err_file, None);
        assert_eq!(info.cpu_percent, None);
        assert_eq!(info.memory_bytes, None);
        assert_eq!(info.dog, None);
        assert_eq!(info.lambs, None);
        assert_eq!(info.last_exit, None);
    }

    /// Every field is given a value distinct from every other field's
    /// default, so a copy-pasted setter body shows up as a mismatch.
    #[test]
    fn every_setter_writes_its_own_field_and_no_other() {
        let built = ProcessInfo::builder(3, "web", ProcStatus::Online)
            .pid(Some(4242))
            .restarts(1)
            .uptime_ms(60_000)
            .fold(Some("backend".to_string()))
            .out_file(Some("/home/ada/.shep/logs/web-0-out.log".to_string()))
            .err_file(Some("/home/ada/.shep/logs/web-0-err.log".to_string()))
            .cpu_percent(Some(12.5))
            .memory_bytes(Some(48 * 1024 * 1024))
            .dog(None)
            .last_exit(Some(ExitInfo {
                code: Some(1),
                signal: None,
            }))
            .max_memory(Some(512 * 1024 * 1024))
            .build();

        // `sample_info()` is a struct literal on purpose: it is the one
        // place that names every field by hand, so this comparison fails the
        // day the struct grows a field the builder cannot set.
        assert_eq!(built, sample_info());

        // `sample_info()`'s `dog` is `None`, the builder's default too, so an
        // empty `dog` setter body would pass the comparison above. It cannot
        // be changed: it feeds pinned snapshots.
        assert_eq!(
            ProcessInfo::builder(1, "metrics", ProcStatus::Online)
                .dog(Some(DogSource::BuiltIn))
                .build()
                .dog,
            Some(DogSource::BuiltIn),
            "an empty `dog` setter body is invisible to the comparison above"
        );

        // `lambs`, on `dog`'s terms above.
        assert_eq!(
            ProcessInfo::builder(1, "web", ProcStatus::Online)
                .lambs(Some(vec![Lamb::new(4243, "node")]))
                .build()
                .lambs,
            Some(vec![Lamb::new(4243, "node")]),
            "an empty `lambs` setter body is invisible to the comparison above"
        );

        // `smit`, on the same terms, and the field a third party writes: an
        // empty setter body drops every dog's mark.
        assert_eq!(
            ProcessInfo::builder(1, "web", ProcStatus::Online)
                .smit(Some("\u{25b2} main@a1b2c3".to_string()))
                .build()
                .smit
                .as_deref(),
            Some("\u{25b2} main@a1b2c3"),
            "an empty `smit` setter body is invisible to the comparison above"
        );

        // `handshook`, on the same terms.
        assert_eq!(
            ProcessInfo::builder(1, "web", ProcStatus::Online)
                .handshook(Some(false))
                .build()
                .handshook,
            Some(false),
            "an empty `handshook` setter body is invisible to the comparison above"
        );

        // `dog_stale`, paired with `handshook`: both default to `None`.
        assert_eq!(
            ProcessInfo::builder(1, "web", ProcStatus::Online)
                .dog_stale(Some(true))
                .build()
                .dog_stale,
            Some(true),
            "an empty `dog_stale` setter body is invisible to the comparison above"
        );

        // `pending`, on the same terms.
        assert_eq!(
            ProcessInfo::builder(1, "web", ProcStatus::Online)
                .pending(Some(vec!["env".to_string()]))
                .build()
                .pending,
            Some(vec!["env".to_string()]),
            "an empty `pending` setter body is invisible to the comparison above"
        );

        // `overridden`, on the same terms.
        assert_eq!(
            ProcessInfo::builder(1, "web", ProcStatus::Online)
                .overridden(Some(vec!["cwd".to_string()]))
                .build()
                .overridden,
            Some(vec!["cwd".to_string()]),
            "an empty `overridden` setter body is invisible to the comparison above"
        );
    }

    #[test]
    fn lambs_distinguishes_not_walked_from_walked_and_empty() {
        let not_walked = ProcessInfo::builder(1, "web", ProcStatus::Online).build();
        assert_eq!(not_walked.lambs, None);

        let walked_empty = ProcessInfo::builder(1, "web", ProcStatus::Online)
            .lambs(Some(Vec::new()))
            .build();
        assert_eq!(walked_empty.lambs, Some(Vec::new()));
    }

    #[test]
    fn a_process_info_without_a_lambs_key_still_deserializes() {
        let fixture = r#"{
            "id": 3, "name": "web", "status": "online", "pid": 4242,
            "restarts": 0, "uptime_ms": 100, "fold": null,
            "out_file": null, "err_file": null,
            "cpu_percent": null, "memory_bytes": null, "dog": null
        }"#;
        let info: ProcessInfo = serde_json::from_str(fixture).unwrap();
        assert_eq!(info.lambs, None);
    }

    /// argv holds credentials (`--password=`, `?token=`) and
    /// `shep describe --format json` is output people paste into issues.
    #[test]
    fn a_lamb_is_a_pid_and_an_executable_name() {
        let lamb = Lamb::new(4243, "node");
        let json = serde_json::to_string(&lamb).unwrap();
        assert_eq!(json, r#"{"pid":4243,"name":"node"}"#);
        assert_eq!(serde_json::from_str::<Lamb>(&json).unwrap(), lamb);
    }

    #[test]
    fn a_dog_source_serializes_snake_case_under_its_kind() {
        assert_eq!(
            serde_json::to_string(&DogSource::BuiltIn).unwrap(),
            r#"{"kind":"built_in"}"#
        );
        let adopted = DogSource::Adopted {
            path: "/usr/local/bin/shep-otel".to_string(),
        };
        let wire = r#"{"kind":"adopted","path":"/usr/local/bin/shep-otel"}"#;
        assert_eq!(serde_json::to_string(&adopted).unwrap(), wire);
        assert_eq!(serde_json::from_str::<DogSource>(wire).unwrap(), adopted);
    }

    #[test]
    fn v1_process_info_without_a_dog_marker_still_deserializes() {
        let fixture = r#"{"id":3,"name":"web","status":"online","pid":4242,"restarts":1,"uptime_ms":60000,"fold":"backend","out_file":"/l/o.log","err_file":"/l/e.log","cpu_percent":12.5,"memory_bytes":50331648}"#;
        let info: ProcessInfo = serde_json::from_str(fixture).unwrap();
        assert_eq!(info.dog, None);
    }

    /// No field here carries `#[serde(default)]`: serde's derive resolves a
    /// missing key to `None` for a field whose type is syntactically
    /// `Option<...>`.
    #[test]
    fn a_process_info_without_a_last_exit_key_still_deserializes() {
        let fixture = r#"{"id":3,"name":"web","status":"online","pid":4242,"restarts":1,"uptime_ms":60000,"fold":"backend","out_file":"/l/o.log","err_file":"/l/e.log","cpu_percent":12.5,"memory_bytes":50331648,"dog":null,"lambs":null}"#;
        let info: ProcessInfo = serde_json::from_str(fixture).unwrap();
        assert_eq!(info.last_exit, None);
    }

    #[test]
    fn a_signal_request_and_its_reply_round_trip() {
        let request = Request::Signal {
            selector: SelectorSpec::Name("web".to_string()),
            signal: "SIGHUP".to_string(),
        };
        let json = serde_json::to_string(&request).unwrap();
        assert_eq!(serde_json::from_str::<Request>(&json).unwrap(), request);

        let reply = Response::Signalled(vec![
            SignalReply {
                id: 1,
                name: "web".to_string(),
                outcome: SignalOutcome::Delivered,
            },
            SignalReply {
                id: 2,
                name: "web".to_string(),
                outcome: SignalOutcome::NotRunning,
            },
            SignalReply {
                id: 3,
                name: "api".to_string(),
                outcome: SignalOutcome::Failed {
                    reason: "no such process".to_string(),
                },
            },
        ]);
        let json = serde_json::to_string(&reply).unwrap();
        assert_eq!(serde_json::from_str::<Response>(&json).unwrap(), reply);
        // The three tags, spelled out: a variant renamed in Rust changes
        // these strings, compiles clean, and breaks a client matching on them.
        assert!(json.contains(r#""kind":"delivered""#), "{json}");
        assert!(json.contains(r#""kind":"not_running""#), "{json}");
        assert!(json.contains(r#""kind":"failed""#), "{json}");
    }

    /// `instances` is a per-app number, so `shep stock /web.*/ 4` could mean
    /// four each or four total.
    #[test]
    fn a_scale_request_names_one_app_and_a_count() {
        let request = Request::Scale {
            name: "web".to_string(),
            count: 4,
        };
        let json = serde_json::to_string(&request).unwrap();
        assert_eq!(serde_json::from_str::<Request>(&json).unwrap(), request);
        assert!(json.contains(r#""kind":"scale""#), "{json}");
        assert!(json.contains(r#""name":"web""#), "{json}");
        // No `selector` key at all: this verb is not one of the
        // selector-taking family.
        assert!(!json.contains("selector"), "{json}");
    }

    #[test]
    fn a_scaled_reply_carries_its_own_tag() {
        let json = serde_json::to_string(&Response::Scaled(vec![])).unwrap();
        assert_eq!(json, r#"{"kind":"scaled","data":[]}"#);
    }

    /// `Add` and `Start` carry byte-identical payloads and differ by their
    /// `kind` alone.
    #[test]
    fn an_add_request_and_its_reply_round_trip() {
        let request = Request::Add {
            apps: vec![AppConfig::minimal("web", "./srv")],
        };
        let json = serde_json::to_string(&request).unwrap();
        assert_eq!(serde_json::from_str::<Request>(&json).unwrap(), request);
        assert!(json.contains(r#""kind":"add""#), "{json}");

        let reply = Response::Added(vec![]);
        let json = serde_json::to_string(&reply).unwrap();
        assert_eq!(serde_json::from_str::<Response>(&json).unwrap(), reply);
        assert!(json.contains(r#""kind":"added""#), "{json}");
    }

    /// `NotWritten`'s reason is the only thing separating "the app is not
    /// reading its stdin" from "the pipe broke".
    #[test]
    fn a_send_line_request_and_its_reply_round_trip() {
        let request = Request::SendLine {
            selector: SelectorSpec::Name("repl".to_string()),
            line: "reload-config".to_string(),
        };
        let json = serde_json::to_string(&request).unwrap();
        assert_eq!(serde_json::from_str::<Request>(&json).unwrap(), request);

        let reply = Response::SentLine(vec![
            LineReply {
                id: 1,
                name: "repl".to_string(),
                outcome: LineOutcome::Sent,
            },
            LineReply {
                id: 2,
                name: "web".to_string(),
                outcome: LineOutcome::NoStdin,
            },
            LineReply {
                id: 3,
                name: "stuck".to_string(),
                outcome: LineOutcome::NotWritten {
                    reason: "the app did not read its stdin within 2s".to_string(),
                },
            },
        ]);
        let json = serde_json::to_string(&reply).unwrap();
        assert_eq!(serde_json::from_str::<Response>(&json).unwrap(), reply);
        assert!(json.contains(r#""kind":"sent""#), "{json}");
        assert!(json.contains(r#""kind":"no_stdin""#), "{json}");
        assert!(json.contains("did not read its stdin"), "{json}");
    }

    #[test]
    fn a_line_carrying_a_newline_is_still_one_field_on_the_wire() {
        let request = Request::SendLine {
            selector: SelectorSpec::All,
            line: "a\nb".to_string(),
        };
        let json = serde_json::to_string(&request).unwrap();
        // Escaped, not literal: the frame stays one JSON object. Refusing
        // it is the daemon's job, not serde's.
        assert!(json.contains(r#""line":"a\nb""#), "{json}");
        assert_eq!(serde_json::from_str::<Request>(&json).unwrap(), request);
    }

    /// Also pins that the newtype protecting this field costs the wire
    /// nothing: a bare string either way, so no fixture and no protocol
    /// version moves for it.
    #[test]
    fn env_value_debug_does_not_leak() {
        let request = Request::SetSheepEnv {
            name: "web".to_string(),
            key: "DATABASE_URL".to_string(),
            value: Some("postgres://user:hunter2@localhost/app".to_string().into()),
        };
        let debug = format!("{request:?}");
        assert!(!debug.contains("hunter2"), "{debug}");
        assert!(debug.contains("EnvValue(<37 bytes>)"), "{debug}");

        let json = serde_json::to_string(&request).unwrap();
        assert!(
            json.contains(r#""value":"postgres://user:hunter2@localhost/app""#),
            "{json}"
        );
        assert_eq!(serde_json::from_str::<Request>(&json).unwrap(), request);
    }

    /// The exact `Debug` string, not the absence of one value: `entries` is
    /// a map of [`EnvValue`], so what is actually under test is that the
    /// nested redaction renders, and a `contains` check would pass just as
    /// well against a map that printed nothing at all.
    #[test]
    fn put_secrets_round_trips_and_hides_its_values() {
        let request = Request::PutSecrets {
            namespace: "vercel".into(),
            environment: "production".into(),
            entries: BTreeMap::from([(
                "API_KEY".to_string(),
                EnvValue::from("sk_live".to_string()),
            )]),
        };
        let encoded = serde_json::to_string(&request).unwrap();
        assert_eq!(serde_json::from_str::<Request>(&encoded).unwrap(), request);
        assert_eq!(
            format!("{request:?}"),
            "PutSecrets { namespace: \"vercel\", environment: \"production\", \
             entries: {\"API_KEY\": EnvValue(<7 bytes>)} }"
        );
    }

    /// IR-41. `EnvValue` is what keeps the derive on `Request` safe, and this
    /// pins that the batch variant actually uses it.
    #[test]
    fn set_sheep_env_batch_debug_does_not_leak() {
        let request = Request::SetSheepEnvBatch {
            name: "web".to_string(),
            entries: BTreeMap::from([(
                "DB_PASSWORD".to_string(),
                EnvValue::from("hunter2".to_string()),
            )]),
            force: false,
            dry_run: true,
        };
        assert_eq!(
            format!("{request:?}"),
            "SetSheepEnvBatch { name: \"web\", entries: {\"DB_PASSWORD\": EnvValue(<7 bytes>)}, \
             force: false, dry_run: true }"
        );
    }

    /// The wire shape, pinned the way every other variant's is.
    #[test]
    fn set_sheep_env_batch_wire_v8() {
        let request = Request::SetSheepEnvBatch {
            name: "web".to_string(),
            entries: BTreeMap::from([("A".to_string(), EnvValue::from("1".to_string()))]),
            force: true,
            dry_run: false,
        };
        let json = serde_json::to_string(&request).unwrap();
        assert_eq!(
            json,
            r#"{"kind":"set_sheep_env_batch","name":"web","entries":{"A":"1"},"force":true,"dry_run":false}"#
        );
        assert_eq!(serde_json::from_str::<Request>(&json).unwrap(), request);
    }

    /// The reply carries key names and never a value.
    #[test]
    fn sheep_env_batch_response_wire_v8() {
        let response = Response::SheepEnvBatch {
            name: "web".to_string(),
            set: vec!["A".to_string()],
            unchanged: vec!["B".to_string()],
            collisions: Vec::new(),
        };
        let json = serde_json::to_string(&response).unwrap();
        assert_eq!(
            json,
            r#"{"kind":"sheep_env_batch","data":{"name":"web","set":["A"],"unchanged":["B"],"collisions":[]}}"#
        );
        assert_eq!(serde_json::from_str::<Response>(&json).unwrap(), response);
    }

    /// Additive, so the version does not move. Guards against a reflexive bump.
    #[test]
    fn the_batch_variant_did_not_move_the_version() {
        assert_eq!(super::super::PROTOCOL_VERSION, 8);
        assert_eq!(super::super::MIN_SUPPORTED, 8);
    }

    /// The pane edits everything else about a sheep, so the config itself
    /// has to travel; `env` is the one map in it that holds secrets, and
    /// the keys travel while the values never do (IR-41).
    #[test]
    fn a_sheep_config_view_never_carries_an_env_value() {
        let mut config = AppConfig::minimal("web", "./srv");
        config
            .env
            .insert("DB_PASS".to_string(), "hunter2".to_string());
        let view = SheepConfigView::new(config, Vec::new(), Vec::new());
        assert!(view.config.env.is_empty());
        assert_eq!(view.env_keys, ["DB_PASS"]);
        let json = serde_json::to_string(&view).unwrap();
        assert!(!json.contains("hunter2"), "{json}");
    }

    /// A `{:?}` on a `Response` reaches it, and `config` holds `args` and
    /// `cwd` as well as the env keys (IR-41).
    #[test]
    fn a_sheep_config_views_debug_is_the_exact_redacted_string() {
        let mut config = AppConfig::minimal("web", "./srv");
        config.env.insert("A".to_string(), "1".to_string());
        let view = SheepConfigView::new(config, vec!["max_restarts".to_string()], Vec::new());
        assert_eq!(
            format!("{view:?}"),
            r#"SheepConfigView { name: "web", env_keys: 1, overridden: 1, pending: 0 }"#
        );
    }

    #[test]
    fn request_wire_snapshots() {
        let requests = vec![
            Envelope {
                id: 1,
                deadline_ms: Some(5000),
                body: Request::Ping,
            },
            Envelope {
                id: 2,
                deadline_ms: None,
                body: Request::ListFlock,
            },
            Envelope {
                id: 3,
                deadline_ms: None,
                body: Request::Stop {
                    selector: SelectorSpec::Name("web".to_string()),
                },
            },
            Envelope {
                id: 4,
                deadline_ms: None,
                body: Request::Start {
                    apps: vec![AppConfig::minimal("web", "./srv")],
                },
            },
            // `All` rather than a named sheep: the selector `shep reopen`
            // sends when given no argument.
            Envelope {
                id: 5,
                deadline_ms: None,
                body: Request::Reopen {
                    selector: SelectorSpec::All,
                },
            },
            // The same selector as the row above, so the two log-plane rows
            // differ by their `kind` and by nothing else.
            Envelope {
                id: 6,
                deadline_ms: None,
                body: Request::Flush {
                    selector: SelectorSpec::All,
                },
            },
            // The same selector as the `stop` row: `reload` under `stop`'s tag
            // shows up here as two identical objects.
            Envelope {
                id: 7,
                deadline_ms: None,
                body: Request::Reload {
                    selector: SelectorSpec::Name("web".to_string()),
                },
            },
            // `action`/`params` match channel.rs's with-params fixture
            // verbatim, so a trigger reads the same at every hop.
            Envelope {
                id: 8,
                deadline_ms: None,
                body: Request::Trigger {
                    selector: SelectorSpec::Name("web".to_string()),
                    action: "set-log-level".to_string(),
                    params: Some("debug".to_string()),
                },
            },
            // A fieldless verb: a bare `{"kind":"..."}` with no `selector` key.
            Envelope {
                id: 9,
                deadline_ms: None,
                body: Request::SaveRoll,
            },
            // Paired with the `save_roll` row: they differ by their `kind` alone.
            Envelope {
                id: 10,
                deadline_ms: None,
                body: Request::Muster,
            },
            // The three dog verbs. `enable_dog` and `disable_dog` differ by
            // their `kind` and by `source` alone.
            Envelope {
                id: 11,
                deadline_ms: None,
                body: Request::DogConfig {
                    name: "bark".to_string(),
                },
            },
            Envelope {
                id: 12,
                deadline_ms: None,
                body: Request::EnableDog {
                    name: "metrics".to_string(),
                    source: DogSource::BuiltIn,
                },
            },
            Envelope {
                id: 13,
                deadline_ms: None,
                body: Request::DisableDog {
                    name: "metrics".to_string(),
                },
            },
            // `Id`, `Regex` and `Fold` are three newtypes the wire tells apart
            // only by their `kind` tag: a `Fold` under `regex`'s tag turns
            // `shep restart fold:api` into a regex match.
            Envelope {
                id: 14,
                deadline_ms: None,
                body: Request::Describe {
                    selector: SelectorSpec::Id(7),
                },
            },
            Envelope {
                id: 15,
                deadline_ms: None,
                body: Request::Describe {
                    selector: SelectorSpec::Regex("^web-".to_string()),
                },
            },
            Envelope {
                id: 16,
                deadline_ms: None,
                body: Request::Describe {
                    selector: SelectorSpec::Fold("api".to_string()),
                },
            },
            // `SIGHUP` rather than `SIGTERM`: the stop ladder already sends
            // TERM, so a TERM fixture could not tell the two frames apart.
            Envelope {
                id: 17,
                deadline_ms: None,
                body: Request::Signal {
                    selector: SelectorSpec::Name("web".to_string()),
                    signal: "SIGHUP".to_string(),
                },
            },
            // The one verb here whose body has no `selector` key.
            Envelope {
                id: 18,
                deadline_ms: None,
                body: Request::Scale {
                    name: "web".to_string(),
                    count: 4,
                },
            },
            // The line carries no terminator on the wire, since the shepherd
            // appends it.
            Envelope {
                id: 19,
                deadline_ms: None,
                body: Request::SendLine {
                    selector: SelectorSpec::All,
                    line: "reload-config".to_string(),
                },
            },
            // Both halves of the `Option` are pinned, a paint and a clear, so a
            // dog author does not have to guess the clear frame's shape.
            Envelope {
                id: 20,
                deadline_ms: None,
                body: Request::SetSmit {
                    sheep: "web".to_string(),
                    smit: Some(
                        "\u{25b2} main@a1b2c3"
                            .parse()
                            .expect("the reference smit is valid"),
                    ),
                },
            },
            Envelope {
                id: 21,
                deadline_ms: None,
                body: Request::SetSmit {
                    sheep: "web".to_string(),
                    smit: None,
                },
            },
            // An empty `apps`: `start`'s row already pins the payload type, so
            // this row's own are the tag and the key the list travels under.
            Envelope {
                id: 22,
                deadline_ms: None,
                body: Request::ConfigDrift { apps: Vec::new() },
            },
            // The only struct-shaped `SelectorSpec` variant, so the only place
            // `"kind":"instance"` and the `slot` key are pinned.
            Envelope {
                id: 23,
                deadline_ms: None,
                body: Request::Restart {
                    selector: SelectorSpec::Instance {
                        name: "web".to_string(),
                        slot: 2,
                    },
                },
            },
            // The one request an older daemon must never be sent: shep-cli
            // gates it on the daemon's crate version.
            Envelope {
                id: 24,
                deadline_ms: None,
                body: Request::HandoverFitness,
            },
            // The second request gated on the daemon's crate version.
            Envelope {
                id: 25,
                deadline_ms: None,
                body: Request::DogStaleness,
            },
            // The only request carrying a `DeclaredApp`: a merge keys on what a
            // document claimed. `declared_env` is non-empty to show it holds
            // env key names and no env value, and `reset` is pinned at a
            // non-default depth.
            Envelope {
                id: 26,
                deadline_ms: None,
                body: Request::ApplyConfig {
                    apps: vec![DeclaredApp {
                        config: AppConfig::minimal("web", "./srv"),
                        declared: ["name", "script"]
                            .iter()
                            .map(|k| (*k).to_string())
                            .collect(),
                        declared_env: ["DATABASE_URL"].iter().map(|k| (*k).to_string()).collect(),
                    }],
                    reset: ResetDepth::Policy,
                },
            },
            // The same app as the `start` row above: the two differ by their
            // `kind` alone, so a mis-tagged `add` shows up as two identical
            // objects.
            Envelope {
                id: 27,
                deadline_ms: None,
                body: Request::Add {
                    apps: vec![AppConfig::minimal("web", "./srv")],
                },
            },
            // The four config-pane requests. `SheepConfig` takes a name
            // rather than a selector, like `Scale` and `SetSmit` above and
            // for their reason: a pane edits one sheep.
            Envelope {
                id: 28,
                deadline_ms: None,
                body: Request::SheepConfig {
                    name: "web".to_string(),
                },
            },
            // `value` is pinned as `Some`, because the `None` spelling is
            // what removes the key, and a reader that guessed the two apart
            // wrongly would delete an operator's env instead of setting it.
            // The value is a placeholder, not a secret: this is the one
            // request in the enum that carries an env value at all, and it
            // travels in one direction only, nothing ever reads it back.
            Envelope {
                id: 29,
                deadline_ms: None,
                body: Request::SetSheepEnv {
                    name: "web".to_string(),
                    key: "DATABASE_URL".to_string(),
                    value: Some("postgres://localhost/app".to_string().into()),
                },
            },
            // `SetSheepEnv`'s twin for everything that is not `env`, and
            // pinned beside it: the two are one letter apart in the tag and
            // a reader that crossed them would write a config field into an
            // env map. `value` is a bare JSON value rather than a string,
            // which is the half a hand-written reader gets wrong: an
            // integer field is an integer here, not `"32"`.
            Envelope {
                id: 30,
                deadline_ms: None,
                body: Request::SetSheepField {
                    name: "web".to_string(),
                    key: "max_restarts".to_string(),
                    value: serde_json::json!(32),
                },
            },
            // The second request carrying a `DogSectionToml`, and pinned
            // beside its reader: `DogConfig` asks for a section and this
            // writes one back, so the two have to agree about the shape a
            // section takes on the wire.
            Envelope {
                id: 31,
                deadline_ms: None,
                body: Request::SetDogConfig {
                    name: "bark".to_string(),
                    toml: "debounce = \"30s\"\n".to_string().into(),
                },
            },
            // The one request a provider dog sends, and the row that pins
            // what `EnvValue` costs the wire: `entries` is a plain object
            // of strings, so a dog written against this fixture in another
            // language needs no newtype of its own.
            Envelope {
                id: 32,
                deadline_ms: None,
                body: Request::PutSecrets {
                    namespace: "vercel".to_string(),
                    environment: "production".to_string(),
                    entries: BTreeMap::from([(
                        "API_KEY".to_string(),
                        EnvValue::from("sk_live_placeholder".to_string()),
                    )]),
                },
            },
            // `SetSheepEnvBatch`'s own doc comment calls this the densest
            // run of secrets on the wire, so it gets two entries rather
            // than one: a single-entry map would not distinguish an object
            // from a map with one key. `force` and `dry_run` are both
            // pinned away from their default so a silent default flip on
            // either field shows up here.
            Envelope {
                id: 33,
                deadline_ms: None,
                body: Request::SetSheepEnvBatch {
                    name: "web".to_string(),
                    entries: BTreeMap::from([
                        (
                            "DATABASE_URL".to_string(),
                            EnvValue::from("postgres://localhost/app".to_string()),
                        ),
                        (
                            "API_KEY".to_string(),
                            EnvValue::from("sk_live_placeholder".to_string()),
                        ),
                    ]),
                    force: true,
                    dry_run: true,
                },
            },
        ];
        insta::assert_json_snapshot!("request_wire_v8", requests);
    }

    #[test]
    fn reply_wire_snapshots() {
        let replies = vec![
            Reply {
                id: 1,
                result: Ok(Response::Pong),
            },
            Reply {
                id: 2,
                result: Ok(Response::Flock(vec![sample_info()])),
            },
            Reply {
                id: 3,
                result: Err(RpcError {
                    code: RpcErrorCode::NotFound,
                    message: "no sheep matches `web`".to_string(),
                    daemon_version: None,
                }),
            },
            // `ActionReply` is not a `ProcessInfo`. `Replied` is the
            // struct-shaped `ActionOutcome` variant and so the one worth
            // pinning.
            Reply {
                id: 4,
                result: Ok(Response::Triggered(vec![ActionReply {
                    id: 3,
                    name: "web".to_string(),
                    outcome: ActionOutcome::Replied {
                        body: "ok".to_string(),
                    },
                }])),
            },
            // The only struct-shaped `Response` variant; every other one is
            // a newtype over a `Vec` or a unit, both proven above.
            Reply {
                id: 5,
                result: Ok(Response::RollSaved {
                    path: "/home/ada/.shep/flock.json".to_string(),
                    apps: 2,
                }),
            },
            // The present `dog` marker; `sample_info()` pins the absent one.
            // `Adopted` because it is the variant carrying a payload.
            Reply {
                id: 6,
                result: Ok(Response::Flock(vec![ProcessInfo {
                    id: 7,
                    name: "otel".to_string(),
                    dog: Some(DogSource::Adopted {
                        path: "/usr/local/bin/shep-otel".to_string(),
                    }),
                    ..sample_info()
                }])),
            },
            // The section crosses the wire as text, never a typed structure.
            Reply {
                id: 7,
                result: Ok(Response::DogSection {
                    toml: "port = 9615\n".to_string().into(),
                }),
            },
            // The only `Response` variant carrying a bare `ProcessInfo`
            // rather than a `Vec`: `enable` starts exactly one dog.
            Reply {
                id: 8,
                result: Ok(Response::DogStarted(ProcessInfo {
                    id: 4,
                    name: "metrics".to_string(),
                    dog: Some(DogSource::BuiltIn),
                    ..sample_info()
                })),
            },
            // Each row below carries the smallest body that shows its wire
            // shape: the tag is what is being pinned. `Deleted` is a
            // `Vec<u32>`; `Subscribed` and `ShuttingDown` carry nothing.
            Reply {
                id: 9,
                result: Ok(Response::Described(vec![])),
            },
            Reply {
                id: 10,
                result: Ok(Response::Started(vec![])),
            },
            Reply {
                id: 11,
                result: Ok(Response::Stopped(vec![])),
            },
            // Both halves populated, as the row below: `refused` is the one
            // field on either variant a walk fills and a single-target
            // request never does.
            Reply {
                id: 12,
                result: Ok(Response::Restarted {
                    accepted: vec![],
                    refused: vec![SheepRefusal::new(
                        "db",
                        "selector matched no registered sheep",
                    )],
                }),
            },
            Reply {
                id: 13,
                result: Ok(Response::Reloading {
                    accepted: vec![],
                    refused: vec![SheepRefusal::new("db", "db is already being reloaded")],
                }),
            },
            Reply {
                id: 14,
                result: Ok(Response::Deleted(vec![7, 8])),
            },
            Reply {
                id: 15,
                result: Ok(Response::Reopened(vec![])),
            },
            Reply {
                id: 16,
                result: Ok(Response::Flushed(vec![])),
            },
            Reply {
                id: 17,
                result: Ok(Response::Mustered(vec![])),
            },
            Reply {
                id: 18,
                result: Ok(Response::Subscribed),
            },
            Reply {
                id: 19,
                result: Ok(Response::ShuttingDown),
            },
            // `Signalled`, mirroring the `Triggered` row: one row per
            // `SignalOutcome` variant, so no tag is left unproven.
            Reply {
                id: 20,
                result: Ok(Response::Signalled(vec![
                    SignalReply {
                        id: 1,
                        name: "web".to_string(),
                        outcome: SignalOutcome::Delivered,
                    },
                    SignalReply {
                        id: 2,
                        name: "web".to_string(),
                        outcome: SignalOutcome::NotRunning,
                    },
                    SignalReply {
                        id: 3,
                        name: "api".to_string(),
                        outcome: SignalOutcome::Failed {
                            reason: "no such process".to_string(),
                        },
                    },
                ])),
            },
            Reply {
                id: 21,
                result: Ok(Response::Scaled(vec![sample_info()])),
            },
            // `SentLine`, mirroring the `Signalled` row: one row per
            // `LineOutcome` variant.
            Reply {
                id: 22,
                result: Ok(Response::SentLine(vec![
                    LineReply {
                        id: 1,
                        name: "repl".to_string(),
                        outcome: LineOutcome::Sent,
                    },
                    LineReply {
                        id: 2,
                        name: "web".to_string(),
                        outcome: LineOutcome::NoStdin,
                    },
                    LineReply {
                        id: 3,
                        name: "stuck".to_string(),
                        outcome: LineOutcome::NotWritten {
                            reason: "the app did not read its stdin within 2s".to_string(),
                        },
                    },
                ])),
            },
            // A walked lamb tree; every other row pins the `null` shape.
            Reply {
                id: 23,
                result: Ok(Response::Described(vec![
                    ProcessInfo::builder(3, "web", ProcStatus::Online)
                        .pid(Some(4242))
                        .lambs(Some(vec![Lamb::new(4243, "node"), Lamb::new(4244, "sh")]))
                        .build(),
                ])),
            },
            // The killed-by-signal shape of `last_exit`; every row above pins
            // the exited-normally one. `SIGTERM`'s raw number, since this
            // crate carries no name for it.
            Reply {
                id: 24,
                result: Ok(Response::Flock(vec![
                    ProcessInfo::builder(5, "worker", ProcStatus::Stopped)
                        .restarts(1)
                        .last_exit(Some(ExitInfo {
                            code: None,
                            signal: Some(15),
                        }))
                        .build(),
                ])),
            },
            // The one row that pins a smit on the wire; `sample_info()` carries
            // none.
            Reply {
                id: 25,
                result: Ok(Response::SmitPainted(vec![
                    ProcessInfo::builder(3, "web", ProcStatus::Online)
                        .pid(Some(4242))
                        .smit(Some("\u{25b2} main@a1b2c3".to_string()))
                        .build(),
                ])),
            },
            // A sheep drifting in one field and a sheep drifting in several.
            // `env` is one of them: the name travels and the value never does.
            Reply {
                id: 26,
                result: Ok(Response::Drifted(vec![
                    SheepDrift::new("web", vec!["cwd".to_string()]),
                    SheepDrift::new(
                        "api",
                        vec!["args".to_string(), "env".to_string(), "script".to_string()],
                    ),
                ])),
            },
            // The present shape of `instance`; every row above pins its absence.
            Reply {
                id: 27,
                result: Ok(Response::Flock(vec![
                    ProcessInfo::builder(9, "web", ProcStatus::Online)
                        .pid(Some(5150))
                        .instance(Some(2))
                        .build(),
                ])),
            },
            // Both shapes of the handover answer; the difference between them
            // is a `null`.
            Reply {
                id: 28,
                result: Ok(Response::HandoverFitness { refusal: None }),
            },
            Reply {
                id: 29,
                result: Ok(Response::HandoverFitness {
                    refusal: Some("sheep 'web' has a shepherd channel".to_string()),
                }),
            },
            // Both lists non-empty and different: the two carry the same wire
            // shape.
            Reply {
                id: 30,
                result: Ok(Response::DogStaleness {
                    stale: vec!["metrics".to_string()],
                    pending: vec!["bark".to_string()],
                }),
            },
            // A dog whose process is up and which has never answered this
            // shepherd. `dog_stale: false` is the silence still being waited
            // out; the row below is the one it has given up on.
            Reply {
                id: 31,
                result: Ok(Response::Flock(vec![
                    ProcessInfo::builder(10, "log-rotate", ProcStatus::Online)
                        .pid(Some(208_341))
                        .dog(Some(DogSource::Adopted {
                            path: "/usr/local/bin/shep-log-rotate".to_string(),
                        }))
                        .handshook(Some(false))
                        .dog_stale(Some(false))
                        .build(),
                ])),
            },
            Reply {
                id: 32,
                result: Ok(Response::Flock(vec![
                    ProcessInfo::builder(10, "log-rotate", ProcStatus::Online)
                        .pid(Some(208_341))
                        .dog(Some(DogSource::Adopted {
                            path: "/usr/local/bin/shep-log-rotate".to_string(),
                        }))
                        .handshook(Some(false))
                        .dog_stale(Some(true))
                        .build(),
                ])),
            },
            // Three entries, one per shape a load produces: applied, pending,
            // refused. `env` is a pending name on purpose: the name travels
            // and the value never does.
            Reply {
                id: 32,
                result: Ok(Response::Applied(vec![
                    SheepApplied::new("web", vec!["max_memory".to_string()], Vec::new(), None),
                    SheepApplied::new(
                        "api",
                        Vec::new(),
                        vec!["args".to_string(), "env".to_string()],
                        None,
                    ),
                    SheepApplied::new(
                        "worker",
                        Vec::new(),
                        Vec::new(),
                        Some("worker is not registered".to_string()),
                    ),
                ])),
            },
            // `Added`'s tag, all a fixture can prove for a `Vec<ProcessInfo>`
            // variant. Down here because every id in this vector is
            // hand-written.
            Reply {
                id: 33,
                result: Ok(Response::Added(vec![])),
            },
            // The config pane's answer, and the row that proves its whole
            // security property: `env` serializes as an empty object while
            // `env_keys` names the key beside it, so an out-of-tree reader
            // learns here that a value never travels (IR-41).
            Reply {
                id: 34,
                result: Ok(Response::SheepConfig(Box::new(SheepConfigView::new(
                    {
                        let mut config = AppConfig::minimal("web", "./srv");
                        config
                            .env
                            .insert("DATABASE_URL".to_string(), "postgres://x".to_string());
                        config
                    },
                    vec!["max_restarts".to_string()],
                    vec!["env".to_string()],
                )))),
            },
            // The three acknowledgements. None echoes what was written:
            // `SheepEnvSet` names the key and not its value, for the reason
            // the row above pins, `SheepFieldSet` does the same and adds
            // the one bit the caller cannot derive, and `DogConfigSet`
            // names the dog and not the section.
            Reply {
                id: 35,
                result: Ok(Response::SheepEnvSet {
                    name: "web".to_string(),
                    key: "DATABASE_URL".to_string(),
                }),
            },
            // `pending` pinned `true`, because `false` is the value a reader
            // that dropped the field entirely would decode by accident, and
            // the two answers send an operator to different places: one
            // says the change is in force, the other says to reload.
            Reply {
                id: 36,
                result: Ok(Response::SheepFieldSet {
                    name: "web".to_string(),
                    key: "script".to_string(),
                    pending: true,
                }),
            },
            Reply {
                id: 37,
                result: Ok(Response::DogConfigSet {
                    name: "bark".to_string(),
                }),
            },
            // A count and not the entries: a dog already knows what it
            // pushed, and echoing the map back would put every value it
            // sent on the wire a second time for no reader (IR-41).
            Reply {
                id: 38,
                result: Ok(Response::SecretsPut { accepted: 2 }),
            },
            // Key names only, on all three lists, and none empty here on
            // purpose: an empty `Vec` serializes the same whether it holds
            // strings or something else, so a reader that guessed the
            // element type wrong would still pass against an empty-list
            // fixture. `collisions` also proves a forced write can name a
            // key in both `set` and `collisions` at once.
            Reply {
                id: 39,
                result: Ok(Response::SheepEnvBatch {
                    name: "web".to_string(),
                    set: vec!["DATABASE_URL".to_string(), "API_KEY".to_string()],
                    unchanged: vec!["LOG_LEVEL".to_string()],
                    collisions: vec!["API_KEY".to_string()],
                }),
            },
        ];
        insta::assert_json_snapshot!("reply_wire_v8", replies);
    }

    /// Asserts on the JSON, not the struct: a `Vec<String>` cannot say which
    /// of the two a string is, so a build carrying a value would typecheck.
    #[test]
    fn a_sheep_applied_carries_names_and_never_values() {
        let applied = SheepApplied::new(
            "web",
            vec!["cwd".to_string()],
            vec!["env".to_string()],
            None,
        );
        let json = serde_json::to_string(&applied).unwrap();
        assert!(json.contains("\"env\""), "the NAME travels: {json}");
        assert!(
            !json.contains("DATABASE_URL"),
            "and no value ever does: {json}"
        );
    }

    #[test]
    fn a_sheep_applied_debug_prints_the_names_it_was_given() {
        let applied = SheepApplied::new("web", vec!["cwd".to_string()], Vec::new(), None);
        assert_eq!(
            format!("{applied:?}"),
            "SheepApplied { name: \"web\", applied: [\"cwd\"], pending: [], refused: None }"
        );
    }

    #[test]
    fn a_process_info_without_a_smit_key_still_deserializes() {
        let fixture = r#"{"id":1,"name":"web","status":"online","pid":42,"restarts":0,"uptime_ms":10,"fold":null,"out_file":null,"err_file":null,"cpu_percent":null,"memory_bytes":null,"dog":null,"lambs":null,"last_exit":null}"#;
        let info: ProcessInfo = serde_json::from_str(fixture).unwrap();
        assert_eq!(info.smit, None);
    }

    /// The fixture is a dog's row, where `None` means "render this as it
    /// rendered before the field existed", never "never handshaken".
    #[test]
    fn a_process_info_without_a_handshook_key_still_deserializes() {
        let fixture = r#"{"id":1,"name":"metrics","status":"online","pid":42,"restarts":0,"uptime_ms":10,"fold":null,"out_file":null,"err_file":null,"cpu_percent":null,"memory_bytes":null,"dog":{"kind":"built_in"},"lambs":null,"last_exit":null,"smit":null,"instance":0}"#;
        let info: ProcessInfo = serde_json::from_str(fixture).unwrap();
        assert_eq!(info.handshook, None);
        assert_eq!(info.dog, Some(DogSource::BuiltIn));
    }

    /// The fixture carries `handshook: false`, the case that matters: `None`
    /// is "no verdict to report", never "it has not given up".
    #[test]
    fn a_process_info_without_a_dog_stale_key_still_deserializes() {
        let fixture = r#"{"id":1,"name":"metrics","status":"online","pid":42,"restarts":0,"uptime_ms":10,"fold":null,"out_file":null,"err_file":null,"cpu_percent":null,"memory_bytes":null,"dog":{"kind":"built_in"},"lambs":null,"last_exit":null,"smit":null,"instance":0,"handshook":false}"#;
        let info: ProcessInfo = serde_json::from_str(fixture).unwrap();
        assert_eq!(info.dog_stale, None);
        assert_eq!(info.handshook, Some(false));
    }

    #[test]
    fn a_process_info_carries_its_memory_ceiling_and_defaults_to_none() {
        let plain = ProcessInfo::builder(1, "web", ProcStatus::Online).build();
        assert_eq!(
            plain.max_memory, None,
            "a sheep with no ceiling reports none"
        );

        let capped = ProcessInfo::builder(2, "hungry", ProcStatus::Online)
            .max_memory(Some(52 * 1024 * 1024))
            .build();
        assert_eq!(capped.max_memory, Some(54_525_952));
    }

    #[test]
    fn an_older_daemons_process_info_still_decodes() {
        // The field is additive, so a payload written before it existed has to
        // decode with the ceiling absent rather than fail the whole envelope.
        let older = r#"{"id":1,"name":"web","status":"online","restarts":0,"uptime_ms":0}"#;
        let info: ProcessInfo = serde_json::from_str(older).expect("an older payload decodes");
        assert_eq!(info.max_memory, None);
    }

    /// A dog written in another language speaks this wire directly and never
    /// runs `FromStr`.
    #[test]
    fn a_smit_is_validated_when_it_is_deserialized_not_only_when_parsed() {
        for bad in [
            r#""\u001b[2Jgone""#.to_string(),                    // an escape
            r#""a\nb""#.to_string(),                             // a newline
            r#""""#.to_string(),                                 // empty
            r#""   ""#.to_string(),                              // whitespace
            format!(r#""{}""#, "x".repeat(Smit::MAX_CHARS + 1)), // too long
        ] {
            assert!(
                serde_json::from_str::<Smit>(&bad).is_err(),
                "a daemon must refuse this on the wire: {bad}"
            );
        }
        assert!(serde_json::from_str::<Smit>(r#""\u25b2 main@a1b2c3""#).is_ok());
    }

    /// The hand-written `Deserialize` agrees with the derived `Serialize`
    /// only while the serialize side stays transparent.
    #[test]
    fn a_smit_travels_as_a_bare_string() {
        let smit: Smit = "\u{25b2} main@a1b2c3".parse().expect("valid");
        let json = serde_json::to_string(&smit).unwrap();
        assert_eq!(json, "\"\u{25b2} main@a1b2c3\"");
        assert_eq!(serde_json::from_str::<Smit>(&json).unwrap(), smit);
    }

    #[test]
    fn a_smit_is_capped_in_characters_not_bytes() {
        let cjk = "\u{7f8a}".repeat(Smit::MAX_CHARS);
        assert_eq!(cjk.len(), Smit::MAX_CHARS * 3);
        assert!(cjk.parse::<Smit>().is_ok(), "{cjk}");
        assert_eq!(
            "x".repeat(Smit::MAX_CHARS + 1).parse::<Smit>(),
            Err(SmitError::TooLong {
                chars: Smit::MAX_CHARS + 1
            })
        );
    }

    #[test]
    fn a_smit_is_stored_exactly_as_it_arrived() {
        let padded: Smit = "  main@a1b2c3  ".parse().expect("valid");
        assert_eq!(padded.as_str(), "  main@a1b2c3  ");
        assert_eq!(padded.to_string(), "  main@a1b2c3  ");
    }

    #[test]
    fn v1_fixture_still_deserializes() {
        // Committed byte fixture from protocol v1. If this breaks, bump
        // PROTOCOL_VERSION and record it in the CHANGELOG.
        let fixture = r#"{"id":7,"deadline_ms":null,"body":{"kind":"stop","selector":{"kind":"name","value":"web"}}}"#;
        let env: Envelope = serde_json::from_str(fixture).unwrap();
        assert_eq!(env.id, 7);
        assert!(matches!(
            env.body,
            Request::Stop { selector: SelectorSpec::Name(ref n) } if n == "web"
        ));
    }

    #[test]
    fn hello_handshake_shape() {
        let hello = Hello {
            client_version: "0.1.0".to_string(),
            protocol: PROTOCOL_VERSION,
            dog_name: None,
        };
        let json = serde_json::to_string(&hello).unwrap();
        assert_eq!(json, r#"{"client_version":"0.1.0","protocol":8}"#);
    }

    #[test]
    fn a_dogs_hello_names_the_dog_and_nothing_elses_does() {
        let dog = Hello {
            client_version: "0.1.0".to_string(),
            protocol: PROTOCOL_VERSION,
            dog_name: Some("metrics".to_string()),
        };
        let json = serde_json::to_string(&dog).unwrap();
        assert_eq!(
            json,
            r#"{"client_version":"0.1.0","protocol":8,"dog_name":"metrics"}"#
        );
        assert_eq!(serde_json::from_str::<Hello>(&json).unwrap(), dog);
    }

    /// `Hello` is the version-negotiation frame, so `deny_unknown_fields`
    /// here would refuse a newer client before `protocol` is read, leaving
    /// neither peer able to report the skew.
    #[test]
    fn a_hello_without_a_dog_name_still_parses() {
        let fixture = r#"{"client_version":"0.1.14","protocol":2}"#;
        let hello: Hello = serde_json::from_str(fixture).unwrap();
        assert_eq!(hello.protocol, 2);
        assert_eq!(hello.dog_name, None);

        // The other direction: an older daemon ignores a key it does not
        // know. `unknown_to_an_older_daemon` stands in for `dog_name`.
        let newer = r#"{"client_version":"9.9.9","protocol":2,"dog_name":"metrics","unknown_to_an_older_daemon":true}"#;
        let hello: Hello = serde_json::from_str(newer).unwrap();
        assert_eq!(hello.protocol, 2);
        assert_eq!(hello.dog_name.as_deref(), Some("metrics"));
    }

    #[test]
    fn hello_ack_handshake_shape() {
        let ack = HelloAck {
            daemon_version: "0.5.0".to_string(),
            protocol: PROTOCOL_VERSION,
            pid: 1234,
            min_supported: Some(crate::protocol::MIN_SUPPORTED),
        };
        let json = serde_json::to_string(&ack).unwrap();
        assert_eq!(
            json,
            r#"{"daemon_version":"0.5.0","protocol":8,"pid":1234,"min_supported":8}"#
        );
        assert_eq!(serde_json::from_str::<HelloAck>(&json).unwrap(), ack);
    }

    /// `min_supported` is `None` from a daemon predating the floor, and the
    /// omission has to be a missing key rather than `null`, or it would move
    /// `PROTOCOL_VERSION` for every daemon that already ships one.
    #[test]
    fn hello_ack_without_min_supported_omits_the_key_not_nulls_it() {
        let ack = HelloAck {
            daemon_version: "0.5.0".to_string(),
            protocol: PROTOCOL_VERSION,
            pid: 1234,
            min_supported: None,
        };
        let json = serde_json::to_string(&ack).unwrap();
        assert_eq!(
            json,
            r#"{"daemon_version":"0.5.0","protocol":8,"pid":1234}"#
        );
        assert!(!json.contains("min_supported"));
    }

    /// An old daemon fixture, from before the floor existed, still decodes.
    #[test]
    fn an_old_hello_ack_without_min_supported_still_parses() {
        let fixture = r#"{"daemon_version":"0.1.14","protocol":2,"pid":9}"#;
        let ack: HelloAck = serde_json::from_str(fixture).unwrap();
        assert_eq!(ack.protocol, 2);
        assert_eq!(ack.min_supported, None);
    }

    #[test]
    fn hello_reply_carries_typed_skew_error() {
        let refusal: HelloReply = Err(RpcError {
            code: RpcErrorCode::ProtocolMismatch,
            message: "daemon speaks protocol 1, client sent 2".to_string(),
            daemon_version: None,
        });
        let json = serde_json::to_string(&refusal).unwrap();
        assert_eq!(
            json,
            r#"{"Err":{"code":"protocol_mismatch","message":"daemon speaks protocol 1, client sent 2"}}"#
        );
        let back: HelloReply = serde_json::from_str(&json).unwrap();
        assert_eq!(back, refusal);
    }

    #[test]
    fn v1_reply_fixture_still_deserializes() {
        // Committed byte fixture, protocol v1.
        let ok = r#"{"id":1,"result":{"Ok":{"kind":"pong"}}}"#;
        let reply: Reply = serde_json::from_str(ok).unwrap();
        assert!(matches!(reply.result, Ok(Response::Pong)));
        let err = r#"{"id":2,"result":{"Err":{"code":"not_found","message":"no sheep"}}}"#;
        let reply: Reply = serde_json::from_str(err).unwrap();
        assert_eq!(reply.result.unwrap_err().code, RpcErrorCode::NotFound);
    }

    #[test]
    fn v1_hello_ack_fixture_still_deserializes() {
        let fixture = r#"{"Ok":{"daemon_version":"0.1.0","protocol":1,"pid":4242}}"#;
        let ack: HelloReply = serde_json::from_str(fixture).unwrap();
        assert_eq!(ack.unwrap().pid, 4242);
    }

    #[test]
    fn v1_process_info_without_stats_still_deserializes() {
        let fixture = r#"{"id":3,"name":"web","status":"online","pid":4242,"restarts":1,"uptime_ms":60000,"fold":"backend","out_file":"/l/o.log","err_file":"/l/e.log"}"#;
        let info: ProcessInfo = serde_json::from_str(fixture).unwrap();
        assert_eq!(info.cpu_percent, None);
        assert_eq!(info.memory_bytes, None);
    }

    #[test]
    fn v1_process_info_without_log_paths_still_deserializes() {
        // Committed byte fixture from before `out_file`/`err_file` existed.
        let fixture = r#"{"id":3,"name":"web","status":"online","pid":4242,"restarts":1,"uptime_ms":60000,"fold":"backend"}"#;
        let info: ProcessInfo = serde_json::from_str(fixture).unwrap();
        assert_eq!(info.id, 3);
        assert_eq!(info.out_file, None);
        assert_eq!(info.err_file, None);
    }

    #[test]
    fn an_old_client_still_decodes_a_new_process_info() {
        // `ProcessInfo` carries no `deny_unknown_fields`, unlike the config
        // types in `crate::config`, so extra keys are ignored.
        #[derive(Deserialize)]
        struct V1ProcessInfo {
            id: u32,
            fold: Option<String>,
        }

        let current = serde_json::to_string(&sample_info()).unwrap();
        let old: V1ProcessInfo = serde_json::from_str(&current).unwrap();
        assert_eq!(old.id, 3);
        assert_eq!(old.fold.as_deref(), Some("backend"));
    }

    #[test]
    fn an_rpc_error_without_a_daemon_version_serializes_exactly_as_before() {
        // `skip_serializing_if` is what makes the field free: no
        // `"daemon_version":null` key for an older client to ignore.
        let plain = RpcError {
            code: RpcErrorCode::NotFound,
            message: "no sheep".to_string(),
            daemon_version: None,
        };
        assert_eq!(
            serde_json::to_string(&plain).unwrap(),
            r#"{"code":"not_found","message":"no sheep"}"#
        );
    }

    #[test]
    fn a_v1_rpc_error_fixture_deserializes_with_no_daemon_version() {
        let fixture =
            r#"{"code":"protocol_mismatch","message":"daemon speaks protocol 1, client sent 2"}"#;
        let err: RpcError = serde_json::from_str(fixture).unwrap();
        assert_eq!(err.code, RpcErrorCode::ProtocolMismatch);
        assert_eq!(err.daemon_version, None);
    }

    #[test]
    fn an_old_client_ignores_an_rpc_error_field_it_has_never_seen() {
        // `RpcError` carries no `deny_unknown_fields`, so an optional field
        // may be added without moving `PROTOCOL_VERSION`.
        #[derive(Deserialize)]
        struct OldRpcError {
            code: RpcErrorCode,
            message: String,
        }

        let current = serde_json::to_string(&RpcError {
            code: RpcErrorCode::ProtocolMismatch,
            message: "daemon speaks protocol 1, client sent 2".to_string(),
            daemon_version: Some("0.1.16".to_string()),
        })
        .unwrap();
        let old: OldRpcError = serde_json::from_str(&current).expect("must tolerate");
        assert_eq!(old.code, RpcErrorCode::ProtocolMismatch);
        assert_eq!(old.message, "daemon speaks protocol 1, client sent 2");
    }

    #[test]
    fn deadline_exceeded_code_serializes_snake_case() {
        assert_eq!(
            serde_json::to_string(&RpcErrorCode::DeadlineExceeded).unwrap(),
            "\"deadline_exceeded\""
        );
        assert_eq!(
            serde_json::from_str::<RpcErrorCode>("\"deadline_exceeded\"").unwrap(),
            RpcErrorCode::DeadlineExceeded
        );
    }

    #[test]
    fn action_outcome_kinds_serialize_snake_case_and_round_trip() {
        // The shared snapshots exercise only `Replied`, the struct-shaped
        // variant.
        let cases = [
            (
                ActionOutcome::Replied {
                    body: "pong".to_string(),
                },
                r#"{"kind":"replied","body":"pong"}"#,
            ),
            (ActionOutcome::NoChannel, r#"{"kind":"no_channel"}"#),
            (ActionOutcome::Skipped, r#"{"kind":"skipped"}"#),
            (ActionOutcome::TimedOut, r#"{"kind":"timed_out"}"#),
        ];
        for (outcome, wire) in cases {
            assert_eq!(
                serde_json::to_string(&outcome).unwrap(),
                wire,
                "{outcome:?}"
            );
            assert_eq!(
                serde_json::from_str::<ActionOutcome>(wire).unwrap(),
                outcome
            );
        }
    }

    #[test]
    fn save_roll_serializes_snake_case_with_its_payload_under_data() {
        assert_eq!(
            serde_json::to_string(&Request::SaveRoll).unwrap(),
            r#"{"kind":"save_roll"}"#
        );
        let reply = Response::RollSaved {
            path: "/tmp/flock.json".to_string(),
            apps: 3,
        };
        let wire = r#"{"kind":"roll_saved","data":{"path":"/tmp/flock.json","apps":3}}"#;
        assert_eq!(serde_json::to_string(&reply).unwrap(), wire);
        assert_eq!(serde_json::from_str::<Response>(wire).unwrap(), reply);
    }

    /// The listing is empty on purpose: `reply_wire_snapshots` pins the row
    /// field by field.
    #[test]
    fn muster_serializes_snake_case_with_its_listing_under_data() {
        assert_eq!(
            serde_json::to_string(&Request::Muster).unwrap(),
            r#"{"kind":"muster"}"#
        );
        let reply = Response::Mustered(Vec::new());
        let wire = r#"{"kind":"mustered","data":[]}"#;
        assert_eq!(serde_json::to_string(&reply).unwrap(), wire);
        assert_eq!(serde_json::from_str::<Response>(wire).unwrap(), reply);
    }

    #[test]
    fn the_dog_verbs_serialize_snake_case_with_their_payloads_under_data() {
        assert_eq!(
            serde_json::to_string(&Request::DogConfig {
                name: "bark".to_string()
            })
            .unwrap(),
            r#"{"kind":"dog_config","name":"bark"}"#
        );
        assert_eq!(
            serde_json::to_string(&Request::DisableDog {
                name: "bark".to_string()
            })
            .unwrap(),
            r#"{"kind":"disable_dog","name":"bark"}"#
        );
        let section = Response::DogSection {
            toml: "port = 9615\n".to_string().into(),
        };
        let wire = r#"{"kind":"dog_section","data":{"toml":"port = 9615\n"}}"#;
        assert_eq!(serde_json::to_string(&section).unwrap(), wire);
        assert_eq!(serde_json::from_str::<Response>(wire).unwrap(), section);
    }

    #[test]
    fn dog_section_toml_debug_does_not_leak() {
        // A dog's section routinely holds webhook credentials. Pinned as an
        // exact string, so a `#[derive(Debug)]` on `DogSectionToml` fails
        // here.
        let toml: DogSectionToml =
            "webhook_url = \"https://discord.com/api/webhooks/1/super-secret-token\"\n"
                .to_string()
                .into();
        assert_eq!(format!("{toml:?}"), "DogSectionToml(<70 bytes>)");

        let response = Response::DogSection { toml };
        assert_eq!(
            format!("{response:?}"),
            "DogSection { toml: DogSectionToml(<70 bytes>) }"
        );
    }

    /// The fixture cannot agree under either candidate order: by id it is
    /// `web/1, api/2, web/0`, by name `api, web, web`. The two `web` rows
    /// are the tiebreak half, seeded out of order.
    #[test]
    fn a_listing_sorts_by_name_then_by_id() {
        let mut listing = vec![
            ProcessInfo::builder(1, "web", ProcStatus::Online).build(),
            ProcessInfo::builder(2, "api", ProcStatus::Online).build(),
            ProcessInfo::builder(0, "web", ProcStatus::Online).build(),
        ];
        sort_flock(&mut listing);

        let seen: Vec<(&str, u32)> = listing
            .iter()
            .map(|info| (info.name.as_str(), info.id))
            .collect();
        assert_eq!(
            seen,
            vec![("api", 2), ("web", 0), ("web", 1)],
            "name first, then id inside a name"
        );
    }

    #[test]
    fn an_instance_slot_survives_a_round_trip_and_defaults_to_absent() {
        let with = ProcessInfo::builder(1, "web", ProcStatus::Online)
            .instance(Some(2))
            .build();
        assert_eq!(with.instance, Some(2));

        let without = ProcessInfo::builder(1, "web", ProcStatus::Online).build();
        assert_eq!(
            without.instance, None,
            "a row nobody set a slot on says so, rather than claiming slot 0"
        );
    }

    #[test]
    fn a_reply_from_a_daemon_without_the_field_deserializes_as_absent() {
        let json = r#"{"id":1,"name":"web","status":"online","pid":null,
            "restarts":0,"uptime_ms":0,"fold":null,"out_file":null,
            "err_file":null,"cpu_percent":null,"memory_bytes":null,"dog":null,
            "lambs":null,"last_exit":null,"smit":null}"#;
        let info: ProcessInfo = serde_json::from_str(json).expect("older reply still parses");
        assert_eq!(info.instance, None);
    }

    #[test]
    fn sort_flock_orders_by_slot_before_id() {
        // A reload gave slot 0 a fresh, higher id. Slot order must still win.
        let mut listing = vec![
            ProcessInfo::builder(9, "web", ProcStatus::Online)
                .instance(Some(0))
                .build(),
            ProcessInfo::builder(2, "web", ProcStatus::Online)
                .instance(Some(1))
                .build(),
        ];
        sort_flock(&mut listing);
        assert_eq!(
            listing.iter().map(|i| i.id).collect::<Vec<_>>(),
            vec![9, 2],
            "slot 0 leads even though its id is higher"
        );
    }

    #[test]
    fn sort_flock_falls_back_to_id_when_no_row_carries_a_slot() {
        let mut listing = vec![
            ProcessInfo::builder(5, "web", ProcStatus::Online).build(),
            ProcessInfo::builder(3, "web", ProcStatus::Online).build(),
        ];
        sort_flock(&mut listing);
        assert_eq!(
            listing.iter().map(|i| i.id).collect::<Vec<_>>(),
            vec![3, 5],
            "an older daemon's listing sorts exactly as it does today"
        );
    }
}