shep-deploy 0.2.9

//! On removal, put every sheep back where it ran before this dog took over.
//!
//! shep runs [`all`] as this dog's own `on-remove` hook, once, before
//! forgetting it. There is no next tick and no retry beyond what is written
//! here, so a failure has to be loud rather than swallowed.
//!
//! # The failure this prevents
//!
//! An operator rehomes the dog, goes back to `~/ReactMap` because that is
//! where they think their app lives, restarts the sheep, and cannot work
//! out why nothing updates. That sheep's `cwd` was never `~/ReactMap`; it
//! was a path under `$SHEP_HOME` they have no reason to know about.
//! [`State::origin_cwd`] and [`State::origin_script`], captured once at
//! opt-in, are where this module puts it back.
//!
//! # Two cases, both answered from `deploy.toml`
//!
//! A sheep that pre-existed the dog has `origin_cwd` and `origin_script`
//! and is restored to them. A sheep the dog bootstrapped has neither, so
//! there is nothing to restore: it is left running from `current`,
//! unchanged, and the report says so plainly. Deleting an app because a
//! deploy tool was uninstalled would be far worse than leaving it.
//!
//! # Why there is a fallback, and why one outcome is worse than the others
//!
//! [`Request::Delete`] is stop plus deregister, and `FlockRegistry::roll`
//! drops a name with no live instance, so delete-then-start is destructive
//! if the start is refused: the sheep would be gone from the flock AND the
//! roll, not returning on a reboot. A refused restore is retried with the
//! config the shepherd had a moment ago, which covers the common causes -
//! a transient refusal, a bad `origin_script`, a `user` that no longer
//! resolves - at the cost of one extra request. Only when that fallback
//! also fails is the sheep genuinely gone, and [`Restored::Lost`] says so
//! in words an operator cannot misread as "still running".
//!
//! [`Request::Delete`]: shep_client::shep_core::protocol::Request::Delete

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};

use shep_client::shep_core::config::AppConfig;

use crate::daemon::Daemon;
use crate::error::Error;
use crate::paths::{self, Tree};
use crate::roll;
use crate::state::State;

/// What became of one sheep during removal.
#[derive(Debug)]
pub enum Restored {
    /// Put back at its own checkout, from before this dog took over.
    Returned {
        /// The sheep restored.
        sheep: String,
        /// Where it was put back to.
        to: PathBuf,
    },
    /// A cutover that never landed, so the sheep was never moved and is
    /// already running where it belongs.
    ///
    /// `optin::prepare` writes `origin_cwd` and `origin_script` and then
    /// stops; `cut_over` is what re-registers the sheep against `current`.
    /// Between the two, a tree exists and names an origin for a sheep that
    /// is still running exactly where it always was. Restoring it would
    /// delete and restart a healthy app to put it back where it already is,
    /// and a `Start` that then failed would leave it stopped.
    NeverMoved {
        /// The sheep that was never moved.
        sheep: String,
        /// Where it has been running the whole time.
        at: PathBuf,
    },
    /// The dog bootstrapped this sheep, so there was nowhere to restore it
    /// to. Left running from `current`, unchanged.
    LeftRunning {
        /// The sheep left running.
        sheep: String,
        /// Where it is still running from.
        from: PathBuf,
    },
    /// Nothing was changed by this dog. This says nothing about whether the
    /// sheep is still registered or running: `all` also uses it for a
    /// `deploy.toml` it could not even read, and `put_back` uses it for a
    /// sheep the shepherd no longer has registered at all.
    Failed {
        /// The sheep that could not be restored.
        sheep: String,
        /// Why.
        why: String,
    },
    /// A directory under the deploy root that could not be read as targets
    /// at all: the root itself, or one entry with a name no sheep can have.
    ///
    /// Not a [`Self::Failed`], whose `sheep` names a sheep; this names a
    /// path, and nothing under it was looked at.
    Unlisted {
        /// The directory.
        dir: PathBuf,
        /// Why.
        why: String,
    },
    /// The restore itself was refused, but the fallback put the shepherd's
    /// own previous config back.
    ///
    /// Distinct from [`Self::Failed`] because it is NOT "left as it is":
    /// the delete already ran and `start(vec![current])` then succeeded, so
    /// the sheep was stopped and a fresh instance started. The registered
    /// config an operator reads in `shep flock` now matches what it was a
    /// moment ago, but nothing mid-flight survived the restart to get
    /// there.
    Reset {
        /// The sheep that could not be restored, but was put back running.
        sheep: String,
        /// Why the restore itself failed.
        why: String,
    },
    /// At least one delete already landed for this sheep before another
    /// one refused, so there is still at least one instance to account
    /// for.
    ///
    /// Its own variant because neither of the other failure shapes is true
    /// of it: `Failed`'s "left as it is" is false - at least one instance
    /// really is gone - and `Lost`'s "gone from the flock" over-claims,
    /// since instances the loop never reached may still be serving. What
    /// this crate cannot know from here is whether the registered config
    /// survived: `Delete` is stop plus deregister, and the roll drops a
    /// name with no live instance, so a delete that lands on the LAST
    /// remaining instance really does deregister it. Only `shep describe`
    /// and `shep flock` can say.
    PartlyDeleted {
        /// The sheep left in this state.
        sheep: String,
        /// Why the delete that stopped partway failed.
        why: String,
    },
    /// The delete succeeded, both the restore and the fallback failed, and
    /// the sheep is now gone from the flock AND from the roll.
    Lost {
        /// The sheep that is gone.
        sheep: String,
        /// Why neither the restore nor the fallback could put it back.
        why: String,
    },
    /// The muster roll itself could not be read, so none of these
    /// pre-existing sheep could be checked against it at all.
    ///
    /// One row for the whole run rather than one per sheep: the roll is a
    /// single read shared by every target, so its failure is a dog-wide
    /// condition, not a property of any one of them. Repeating "sheep is no
    /// longer registered" once per target would also bury the real cause -
    /// [`roll::read`](crate::roll)'s own crafted, actionable message about
    /// a shepherd newer than this dog - behind a guess this module made up
    /// because it never got to check.
    RollUnreadable {
        /// Every pre-existing sheep this run could not check.
        sheep: Vec<String>,
        /// [`roll::registered`](crate::roll)'s own failure.
        why: String,
    },
}

/// Puts every target back, and answers with one row per target.
///
/// Never returns a `Result`, and that is the contract rather than a
/// convenience: an operator asking to remove something is entitled to have
/// it removed, so a failure here becomes a row in the report and the
/// process still exits 0. A dog that refused to be uninstalled because one
/// of five sheep would not restart would be worse than one that did nothing
/// at all.
pub async fn all<D: Daemon>(daemon: &D, shep_home: &Path) -> Vec<Restored> {
    let found = match paths::targets(shep_home) {
        Ok(found) => found,
        // The deploy directory exists but could not be listed. There is
        // nothing to iterate, but silence here reads as success: `report(&[])`
        // is empty, so `on_remove` prints nothing and exits 0 with every
        // sheep under the tree left running. One row naming the directory
        // says something instead.
        Err(err) => {
            return vec![Restored::Unlisted {
                dir: shep_home.join("deploy"),
                why: err.to_string(),
            }];
        }
    };
    let names = found.named;
    // Read once for every target rather than once per target: it costs a
    // SaveRoll round trip, and a removal is not the moment to make N of
    // them. Kept as a `Result` rather than `.unwrap_or_default()`: an empty
    // map here is indistinguishable from "nothing is registered", and
    // `put_back` would report every pre-existing sheep as "no longer
    // registered" - a fabricated cause standing in for whatever
    // `roll::registered` actually failed with.
    let registered = roll::registered(daemon).await;

    let mut results = Vec::new();
    // A target the dog cannot name cannot be restored either, and a sheep
    // may still be running out of it.
    for dir in found.unnamed {
        results.push(Restored::Unlisted {
            why: "its directory's name cannot be a sheep's, so it was never polled and is not \
                  restored. A sheep may still be running from inside it: `shep flock` lists \
                  every sheep and `shep describe <sheep>` its cwd. Rename the directory once \
                  none does, then restore by hand"
                .to_owned(),
            dir,
        });
    }
    // Pre-existing sheep whose restore needs the roll, deferred here rather
    // than reported as they are found: a roll that cannot be read is one
    // failure shared by every one of them, not N separate ones.
    let mut blocked_by_roll = Vec::new();

    for sheep in names {
        let tree = Tree::for_sheep(shep_home, &sheep);
        let state = match State::read(&tree.state_file()) {
            Ok(state) => state,
            Err(err) => {
                results.push(Restored::Failed {
                    sheep,
                    why: err.to_string(),
                });
                continue;
            }
        };

        // Nothing to restore means the dog bootstrapped this sheep, so it
        // is left running and TOLD about. Deleting an app because a deploy
        // tool was uninstalled would be much worse than leaving it. This
        // needs no roll at all, so a roll that failed to read does not
        // touch it.
        // Read before the origin fields move out below.
        let never_moved = state.deployed.is_none();
        let origin = state.origin;
        let (Some(cwd), Some(script)) = (state.origin_cwd, state.origin_script) else {
            results.push(Restored::LeftRunning {
                sheep,
                from: tree.current(),
            });
            continue;
        };

        let Ok(registered) = &registered else {
            blocked_by_roll.push(sheep);
            continue;
        };

        // An origin recorded is not the same as a sheep moved. `prepare`
        // writes both origin fields, and only a landed cutover writes
        // `deployed`, so an absent `deployed` means the cutover did not land.
        //
        // It does NOT mean the cutover never ran, which is why the shepherd is
        // asked as well. `cut_over` writes `deployed` only on the fully
        // verified path; its `NotVerified` and `Failed` arms return without
        // touching it, having already run `undo_start`, whose own repair can
        // fail and leave a newcomer registered. Reading `deployed` alone
        // called that sheep untouched and reported it as running where it
        // belongs, at the one moment this module exists to catch exactly that.
        //
        // `cwd` is the signal because `cut_over` is the only thing that ever
        // sets it to the tree's `current`, so the shepherd's own registration
        // says whether the cutover got as far as registering.
        if never_moved && !registered_against(registered, &sheep, &tree) {
            results.push(Restored::NeverMoved { sheep, at: cwd });
            continue;
        }

        // The report names where the sheep went: the origin's own cwd when
        // the record carries one, the legacy field otherwise, which is the
        // same choice `put_back` makes.
        let to = origin
            .as_ref()
            .and_then(|origin| origin.cwd.as_deref())
            .map_or_else(|| cwd.clone(), PathBuf::from);
        results.push(
            match put_back(daemon, &sheep, registered, origin.as_ref(), &cwd, &script).await {
                PutBack::Done => Restored::Returned { sheep, to },
                PutBack::Untouched(err) => Restored::Failed {
                    sheep,
                    why: err.to_string(),
                },
                PutBack::Reset(err) => Restored::Reset {
                    sheep,
                    why: err.to_string(),
                },
                PutBack::PartlyDeleted(err) => Restored::PartlyDeleted {
                    sheep,
                    why: err.to_string(),
                },
                PutBack::Deleted(err) => Restored::Lost {
                    sheep,
                    why: err.to_string(),
                },
            },
        );
    }

    if let Err(err) = &registered
        && !blocked_by_roll.is_empty()
    {
        results.push(Restored::RollUnreadable {
            sheep: blocked_by_roll,
            why: err.to_string(),
        });
    }

    results
}

/// What happened to one sheep, distinguishing "nothing changed" from
/// "it is deleted", because those need different words in the report.
enum PutBack {
    /// Re-registered at its own checkout.
    Done,
    /// Nothing happened: the restore was never attempted, or was refused
    /// before anything about the sheep changed.
    Untouched(Error),
    /// The restore was refused, but the fallback re-registered the config
    /// the shepherd already had. NOT the same claim as `Untouched`: the
    /// delete already ran, so the sheep was stopped and a fresh instance
    /// started to get back to that config.
    Reset(Error),
    /// At least one delete already landed before another one refused.
    /// NOT `Untouched` (at least one instance really is gone) and NOT
    /// `Deleted` either - whether the registered config survived depends
    /// on whether the deletes that landed included the last live instance,
    /// which this loop does not track.
    PartlyDeleted(Error),
    /// The delete landed and neither the restore nor the fallback did.
    Deleted(Error),
}

/// Re-registers one sheep as it was before this dog took over: the whole
/// definition when the record carries one, or the `cwd` and `script` it ran
/// with over the shepherd's current definition when it does not.
///
/// Delete THEN start, and the order is tested. `Request::Start` on an
/// already-registered name adds an instance rather than re-registering it,
/// so starting first would leave the sheep running from both places at
/// once, which is the same fact the cutover is built on. Here that order
/// also leaves a CLEAN roll, unlike the cutover's: the delete drops the
/// name, so the following `Start` re-records against a name with no stale
/// entry behind it.
///
/// # Why there is a fallback
///
/// `Delete` is stop plus deregister, and the roll drops a name with no live
/// instance, so a refused `Start` here leaves the sheep gone from the flock
/// AND the roll, not returning on a reboot. The fallback re-registers the
/// config the shepherd had a moment ago, which costs one request on the
/// transient failures that are the common case. Only when that fails too is
/// the sheep genuinely gone, and the caller says so in those words.
async fn put_back<D: Daemon>(
    daemon: &D,
    sheep: &str,
    registered: &BTreeMap<String, AppConfig>,
    origin: Option<&AppConfig>,
    cwd: &Path,
    script: &str,
) -> PutBack {
    let Some(current) = registered.get(sheep).cloned() else {
        return PutBack::Untouched(Error::Config(format!(
            "{sheep} is no longer registered, so there is nothing to put back"
        )));
    };

    // The whole pre-adoption definition when the record has it, so `env`,
    // `instances`, the probes and the rest go back too. A record from before
    // that field carries only `cwd` and `script`, and those go over whatever
    // the shepherd has now, which is the deployed repository's definition.
    let restored = match origin {
        Some(origin) => {
            let mut restored = origin.clone();
            restored.name = sheep.to_owned();
            // `prepare` refuses a sheep with no cwd, so an origin without
            // one is a hand-edited record. The legacy field is the next
            // best witness, and using it here is what keeps the report,
            // which names the same field, truthful.
            restored
                .cwd
                .get_or_insert_with(|| cwd.display().to_string());
            restored
        }
        None => {
            let mut restored = current.clone();
            restored.cwd = Some(cwd.display().to_string());
            restored.script = script.to_owned();
            restored
        }
    };

    let live = match daemon.describe(sheep).await {
        Ok(live) => live,
        Err(err) => return PutBack::Untouched(err),
    };
    let mut any_delete_landed = false;
    for info in &live {
        if let Err(err) = daemon.delete(info.id).await {
            // The discriminator is whether a PRIOR delete already landed,
            // not whether this one failed: a refusal on the very first
            // instance with nothing behind it has changed nothing at all,
            // and is `Untouched` exactly as before this variant existed. A
            // refusal after at least one delete succeeded is the case
            // `Untouched` cannot describe truthfully - see
            // `PutBack::PartlyDeleted`'s doc.
            return if any_delete_landed {
                PutBack::PartlyDeleted(err)
            } else {
                PutBack::Untouched(err)
            };
        }
        any_delete_landed = true;
    }

    match daemon.start(vec![restored]).await {
        Ok(_) => PutBack::Done,
        Err(err) => {
            // The sheep is deregistered at this point. Put the shepherd's
            // own config back rather than leaving it deleted, because a
            // refused restore is usually a bad origin_script or a user that
            // no longer resolves, and the config that was working a moment
            // ago still is.
            if daemon.start(vec![current]).await.is_ok() {
                PutBack::Reset(err)
            } else {
                PutBack::Deleted(err)
            }
        }
    }
}

/// The report shep's hook pipes to the operator, which is the whole of what
/// they see about this.
#[must_use]
pub fn report(results: &[Restored]) -> String {
    if results.is_empty() {
        // Silence here is indistinguishable from success. There were no
        // deploy targets at all, and the operator is entitled to be told
        // that rather than left to guess why nothing printed.
        return "no deploy targets, nothing to restore\n".to_owned();
    }
    results
        .iter()
        .map(|result| match result {
            Restored::Returned { sheep, to } => {
                format!("{sheep} restored to {}\n", to.display())
            }
            // Rin's condition for accepting the leave-running case at all.
            // Without this line, "left running, unchanged" is
            // indistinguishable from "quietly abandoned somewhere you will
            // not think to look", which is the failure this whole module
            // exists to prevent.
            Restored::NeverMoved { sheep, at } => {
                format!(
                    "{sheep} was never moved - its cutover did not land - and is still running \
                     from {}\n",
                    at.display()
                )
            }
            Restored::LeftRunning { sheep, from } => {
                format!("{sheep} still running from {}\n", from.display())
            }
            Restored::Failed { sheep, why } => {
                format!("{sheep} could not be restored ({why}); nothing was changed by this dog\n")
            }
            Restored::Unlisted { dir, why } => format!(
                "{} could not be listed ({why}); nothing under it was changed by this dog\n",
                crate::shared::printable(dir.display())
            ),
            // NOT the same wording as `Failed`: this sheep was stopped and
            // a fresh instance started to get its own previous config back,
            // so nothing mid-flight survived even though the config an
            // operator reads now matches what it was a moment ago.
            Restored::Reset { sheep, why } => format!(
                "{sheep} could not be restored ({why}), so its previous configuration was put \
                 back instead - doing that stopped it and started it again, so it is running \
                 with the same config as before but nothing mid-flight survived\n"
            ),
            // Neither "left as it is" nor "gone from the flock" is true
            // here, so this gets wording that says exactly that, and sends
            // the operator to the one place that can say what is really
            // running: `shep describe`.
            Restored::PartlyDeleted { sheep, why } => format!(
                "{sheep}: only SOME of its instances could be stopped before a delete failed \
                 ({why}), so it is neither fully running nor fully removed. Run `shep describe \
                 {sheep}` to see what is actually still there, and `shep flock` for whether it \
                 is still registered, before assuming either.\n"
            ),
            // The row an operator must not misread. Every other outcome
            // leaves a running app; this one does not, and "could not be
            // restored" would have them assume it did.
            Restored::Lost { sheep, why } => format!(
                "{sheep} IS NO LONGER REGISTERED: restoring it failed ({why}) and so did \
                 putting its previous configuration back, so it is stopped and gone from the \
                 flock. It will not come back on its own after a restart. Re-register it from \
                 its own Flockfile.\n"
            ),
            // One row naming every affected sheep and the roll's own
            // cause, rather than one wrong-shaped guess per sheep.
            Restored::RollUnreadable { sheep, why } => format!(
                "{}: none of these could be checked against the muster roll, so none of them \
                 could be restored ({why})\n",
                sheep.join(", ")
            ),
        })
        .collect()
}
/// Whether the shepherd has `sheep` registered against `tree` rather than
/// against the operator's own checkout.
///
/// `crate::optin::cut_over` sets `cwd` to the tree's `current` symlink and
/// nothing else in this crate does, so this is the shepherd's own answer to
/// "did the cutover get as far as registering". A tree's record cannot answer
/// it: `deployed` is written on the verified path only.
fn registered_against(registered: &BTreeMap<String, AppConfig>, sheep: &str, tree: &Tree) -> bool {
    registered
        .get(sheep)
        .and_then(|app| app.cwd.as_deref())
        // Two spellings of one directory compare equal, and a `current`
        // whose release is already gone still compares by its parent: see
        // `shared::resolved` for why a literal comparison here left a
        // sheep the cutover DID register running from the tree.
        .is_some_and(|cwd| crate::shared::same_path(Path::new(cwd), &tree.current()))
}

#[cfg(test)]
mod tests {
    /// fails if a `deploy.toml` that cannot be parsed is skipped silently.
    ///
    /// This is the one moment the record matters most: removal is when the dog
    /// puts every sheep back where its operator will look for it, and a target
    /// it cannot read is a target it cannot restore. Skipping it would leave
    /// an app running from a path under `$SHEP_HOME` with nothing said about
    /// it.
    ///
    /// `poll.rs` pins the identical shape for `tick` in
    /// `a_record_that_cannot_be_read_is_reported_rather_than_skipped`; the
    /// branch here had no equivalent, so a change to `State::read`'s error
    /// path could have gone quiet on one side and been caught only on the
    /// other.
    #[tokio::test]
    async fn a_record_that_cannot_be_read_is_reported_rather_than_skipped() {
        let home = tempfile::tempdir().expect("tempdir");
        let tree = Tree::for_sheep(home.path(), "garbled");
        std::fs::create_dir_all(tree.root()).expect("create the tree");
        std::fs::write(tree.state_file(), "this is not toml").expect("write deploy.toml");

        let results = all(&Recording::new(&[], Refuse::Never), home.path()).await;

        assert_eq!(results.len(), 1, "the target must be reported, not skipped");
        assert!(
            matches!(&results[0], Restored::Failed { sheep, .. } if sheep == "garbled"),
            "an unreadable record must be a reported failure, got: {:?}",
            results[0]
        );
    }

    use std::cell::{Cell, RefCell};
    use std::fs;

    use shep_client::RequestError;
    use shep_client::shep_core::protocol::{ProcessInfo, RpcError, RpcErrorCode};
    use shep_client::shep_core::status::ProcStatus;

    use super::*;
    use crate::fixtures;

    /// Writes a `deploy.toml` for `sheep` recording `origin_cwd` and
    /// `origin_script`, as opt-in would have.
    fn write_target_with_origin(home: &Path, sheep: &str, origin_cwd: &str, origin_script: &str) {
        let tree = Tree::for_sheep(home, sheep);
        fs::create_dir_all(tree.state_file().parent().expect("has a parent"))
            .expect("create target dir");
        let state = State {
            deployed: Some(fixtures::OTHER_SHA.to_owned()),
            origin_cwd: Some(PathBuf::from(origin_cwd)),
            origin_script: Some(origin_script.to_owned()),
            checkout: PathBuf::from(origin_cwd),
            ..fixtures::state()
        };
        state.write(&tree.state_file()).expect("write state");
    }

    /// Writes a `deploy.toml` for `sheep` carrying the whole pre-adoption
    /// definition, as `prepare` writes since 2026-09-04.
    fn write_target_with_full_origin(home: &Path, sheep: &str, origin: &AppConfig) {
        let tree = Tree::for_sheep(home, sheep);
        fs::create_dir_all(tree.state_file().parent().expect("has a parent"))
            .expect("create target dir");
        let cwd = origin.cwd.clone().expect("an origin with a cwd");
        let state = State {
            deployed: Some(fixtures::OTHER_SHA.to_owned()),
            origin_cwd: Some(PathBuf::from(&cwd)),
            origin_script: Some(origin.script.clone()),
            checkout: PathBuf::from(&cwd),
            origin: Some(origin.clone()),
            ..fixtures::state()
        };
        state.write(&tree.state_file()).expect("write state");
    }

    /// fails if removal puts back only `cwd` and `script`. The record
    /// carries the app as the shepherd had it before adoption; a restore
    /// from the shepherd's current definition kept the deployed
    /// repository's `env`, `instances` and probes past the dog's removal.
    #[tokio::test]
    async fn a_sheep_is_restored_to_its_whole_pre_adoption_definition() {
        let home = tempfile::tempdir().expect("tempdir");
        let origin: AppConfig = toml::from_str(
            "name = \"web\"\nscript = \"server.js\"\ncwd = \"/srv/web\"\ninstances = 3\n\
             [env]\nPORT = \"8080\"\n",
        )
        .expect("an app");
        write_target_with_full_origin(home.path(), "web", &origin);
        let daemon = Recording::new(&["web"], Refuse::Never);

        let results = all(&daemon, home.path()).await;

        assert!(
            matches!(&results[0], Restored::Returned { .. }),
            "{results:?}"
        );
        let started = daemon.started();
        let put_back = started.first().expect("one Start");
        assert_eq!(put_back.instances, 3, "instances come back");
        assert_eq!(
            put_back.env.get("PORT").map(String::as_str),
            Some("8080"),
            "env comes back"
        );
        assert_eq!(put_back.cwd.as_deref(), Some("/srv/web"));
        assert_eq!(put_back.script, "server.js");
    }

    /// fails if the report names a directory the sheep was not put back in.
    /// `put_back` starts the origin's own `cwd` when the record carries an
    /// origin, and `origin_cwd` is a hand-editable legacy field that can
    /// disagree with it; the report has to follow the same choice.
    #[tokio::test]
    async fn the_report_names_the_origins_cwd_when_the_record_carries_one() {
        let home = tempfile::tempdir().expect("tempdir");
        let origin: AppConfig =
            toml::from_str("name = \"web\"\nscript = \"server.js\"\ncwd = \"/srv/web\"\n")
                .expect("an app");
        let tree = Tree::for_sheep(home.path(), "web");
        fs::create_dir_all(tree.state_file().parent().expect("has a parent"))
            .expect("create target dir");
        let state = State {
            deployed: Some(fixtures::OTHER_SHA.to_owned()),
            origin_cwd: Some(PathBuf::from("/srv/stale")),
            origin_script: Some("stale.js".to_owned()),
            checkout: PathBuf::from("/srv/web"),
            origin: Some(origin),
            ..fixtures::state()
        };
        state.write(&tree.state_file()).expect("write state");
        let daemon = Recording::new(&["web"], Refuse::Never);

        let results = all(&daemon, home.path()).await;

        assert!(
            matches!(&results[0], Restored::Returned { to, .. } if to == Path::new("/srv/web")),
            "{results:?}"
        );
        let started = daemon.started();
        assert_eq!(started[0].cwd.as_deref(), Some("/srv/web"));
        assert_eq!(started[0].script, "server.js");
    }

    /// fails if an origin with no `cwd` is put back with none while the
    /// report names the legacy field. `prepare` refuses a sheep without a
    /// cwd, so this is a hand-edited record; the legacy field is the next
    /// witness, and what is started and what is reported have to agree.
    #[tokio::test]
    async fn an_origin_without_a_cwd_is_put_back_at_the_legacy_one() {
        let home = tempfile::tempdir().expect("tempdir");
        let origin: AppConfig =
            toml::from_str("name = \"web\"\nscript = \"server.js\"\n").expect("an app");
        assert!(origin.cwd.is_none());
        let tree = Tree::for_sheep(home.path(), "web");
        fs::create_dir_all(tree.state_file().parent().expect("has a parent"))
            .expect("create target dir");
        let state = State {
            deployed: Some(fixtures::OTHER_SHA.to_owned()),
            origin_cwd: Some(PathBuf::from("/srv/web")),
            origin_script: Some("server.js".to_owned()),
            checkout: PathBuf::from("/srv/web"),
            origin: Some(origin),
            ..fixtures::state()
        };
        state.write(&tree.state_file()).expect("write state");
        let daemon = Recording::new(&["web"], Refuse::Never);

        let results = all(&daemon, home.path()).await;

        assert!(
            matches!(&results[0], Restored::Returned { to, .. } if to == Path::new("/srv/web")),
            "{results:?}"
        );
        assert_eq!(daemon.started()[0].cwd.as_deref(), Some("/srv/web"));
    }

    /// Writes a `deploy.toml` for `sheep` with no `origin_cwd` or
    /// `origin_script`, as a dog-bootstrapped sheep has.
    fn write_target_with_origin_absent(home: &Path, sheep: &str) {
        let tree = Tree::for_sheep(home, sheep);
        fs::create_dir_all(tree.state_file().parent().expect("has a parent"))
            .expect("create target dir");
        let state = State {
            deployed: Some(fixtures::OTHER_SHA.to_owned()),
            checkout: PathBuf::from("/srv/deploy-tree"),
            ..fixtures::state()
        };
        state.write(&tree.state_file()).expect("write state");
    }

    /// How many of a `Recording`'s `start` calls get refused.
    enum Refuse {
        Never,
        FirstOnly,
        Always,
    }

    /// A [`Daemon`] double naming a fixed set of already-registered sheep.
    ///
    /// `save_roll` answers with a roll naming every sheep it was
    /// constructed with, `cwd` under the deploy tree and `script` an
    /// arbitrary placeholder - what the shepherd is presumed to have had
    /// registered before this dog's removal began - unless it was built
    /// [`Self::with_unreadable_roll`], which writes a roll this crate's
    /// `shep-core` refuses, so `roll::registered` fails with its own real,
    /// crafted cause rather than this double inventing one.
    ///
    /// `describe` answers [`Self::instances`] running instances for
    /// whichever name is asked, refusing outright if
    /// [`Self::describe_fails`]. `delete` succeeds until the
    /// [`Self::delete_fails_at`]th call, then refuses every one after -
    /// `Some(1)` refuses the very first delete; a higher number lets some
    /// instances go before one fails partway through.
    ///
    /// `calls()` records only `delete` and `start` calls that actually
    /// landed, in order: those are the two that change the flock, and the
    /// ordering this pins is between them. `save_roll` and `describe` are
    /// reads and are not recorded, so a reordering of those does not break
    /// the assertion.
    struct Recording {
        sheep: Vec<&'static str>,
        refuse: Refuse,
        instances: u32,
        describe_fails: bool,
        delete_fails_at: Option<u32>,
        unreadable_roll: bool,
        /// What the roll says each sheep's `cwd` is, when the default fixed
        /// path will not do. A cutover that registered points the sheep at its
        /// tree's own `current`, which only the test knows the path of.
        registered_cwd: Option<String>,
        calls: RefCell<Vec<&'static str>>,
        starts: RefCell<Vec<AppConfig>>,
        attempts: Cell<usize>,
        delete_attempts: Cell<u32>,
    }

    impl Recording {
        fn new(sheep: &[&'static str], refuse: Refuse) -> Self {
            Self {
                sheep: sheep.to_vec(),
                refuse,
                instances: 1,
                describe_fails: false,
                delete_fails_at: None,
                unreadable_roll: false,
                registered_cwd: None,
                calls: RefCell::new(Vec::new()),
                starts: RefCell::new(Vec::new()),
                attempts: Cell::new(0),
                delete_attempts: Cell::new(0),
            }
        }

        /// Every registered sheep accepts every start.
        fn with_registered(sheep: &[&'static str]) -> Self {
            Self::new(sheep, Refuse::Never)
        }

        /// The same, with every sheep registered against `cwd` rather than
        /// the fixed path the roll otherwise reports.
        fn registered_at(mut self, cwd: &std::path::Path) -> Self {
            self.registered_cwd = Some(cwd.display().to_string());
            self
        }

        /// The very first `start` call, across every sheep, is refused;
        /// every later one is accepted.
        fn refusing_first_start_only(sheep: &[&'static str]) -> Self {
            Self::new(sheep, Refuse::FirstOnly)
        }

        /// Every `start` call is refused, for every sheep.
        fn refusing_every_start(sheep: &[&'static str]) -> Self {
            Self::new(sheep, Refuse::Always)
        }

        /// `save_roll` answers with a roll `roll::registered` cannot parse,
        /// naming no particular sheep - a roll failure is dog-wide, so
        /// nothing here is keyed to the sheep the test writes to disk.
        fn with_unreadable_roll() -> Self {
            let mut this = Self::new(&[], Refuse::Never);
            this.unreadable_roll = true;
            this
        }

        /// `describe` refuses outright, for every sheep. `put_back` cannot
        /// learn what to delete without it, so nothing is ever deleted or
        /// started.
        fn refusing_describe(sheep: &[&'static str]) -> Self {
            let mut this = Self::new(sheep, Refuse::Never);
            this.describe_fails = true;
            this
        }

        /// A single instance is described, and its only `delete` refuses.
        /// Nothing lands before the failure, so the sheep is fully running
        /// and fully registered - the ordinary shape a refused delete
        /// takes, and the one that must stay `Failed` rather than
        /// `PartlyDeleted`, which needs a delete to have already landed.
        fn refusing_first_delete_only(sheep: &[&'static str]) -> Self {
            let mut this = Self::new(sheep, Refuse::Never);
            this.delete_fails_at = Some(1);
            this
        }

        /// Two instances are described; the first `delete` lands, the
        /// second refuses. This is the only shape any test here gives more
        /// than one instance, deliberately: it is what exercises the branch
        /// where a delete fails PARTWAY through, with at least one instance
        /// already gone and another one still to account for.
        fn refusing_delete_partway(sheep: &[&'static str]) -> Self {
            let mut this = Self::new(sheep, Refuse::Never);
            this.instances = 2;
            this.delete_fails_at = Some(2);
            this
        }

        fn calls(&self) -> Vec<&'static str> {
            self.calls.borrow().clone()
        }

        fn started(&self) -> Vec<AppConfig> {
            self.starts.borrow().clone()
        }
    }

    impl Daemon for Recording {
        async fn describe(&self, sheep: &str) -> Result<Vec<ProcessInfo>, Error> {
            if self.describe_fails {
                return Err(Error::Request(RequestError::Rpc(RpcError {
                    code: RpcErrorCode::Internal,
                    message: "describe refused".to_owned(),
                    daemon_version: None,
                })));
            }
            Ok((0..self.instances)
                .map(|offset| {
                    ProcessInfo::builder(offset + 1, sheep, ProcStatus::Online)
                        .pid(Some(1000 + offset))
                        .build()
                })
                .collect())
        }
        async fn start(&self, apps: Vec<AppConfig>) -> Result<Vec<u32>, Error> {
            let attempt = self.attempts.get();
            self.attempts.set(attempt + 1);
            let refused = match self.refuse {
                Refuse::Never => false,
                Refuse::FirstOnly => attempt == 0,
                Refuse::Always => true,
            };
            // Recorded whether accepted or refused: `started()` is what a
            // test asserts the config a call was ATTEMPTED with, including
            // the refused restore itself.
            self.starts.borrow_mut().extend(apps);
            if refused {
                return Err(Error::Request(RequestError::Rpc(RpcError {
                    code: RpcErrorCode::Internal,
                    message: "refused".to_owned(),
                    daemon_version: None,
                })));
            }
            // `calls()` tracks only what actually changed the flock, so a
            // refused start does not appear here.
            self.calls.borrow_mut().push("start");
            Ok(Vec::new())
        }
        async fn delete(&self, _id: u32) -> Result<(), Error> {
            let attempt = self.delete_attempts.get() + 1;
            self.delete_attempts.set(attempt);
            if self.delete_fails_at == Some(attempt) {
                return Err(Error::Request(RequestError::Rpc(RpcError {
                    code: RpcErrorCode::Internal,
                    message: "delete refused".to_owned(),
                    daemon_version: None,
                })));
            }
            self.calls.borrow_mut().push("delete");
            Ok(())
        }
        async fn save_roll(&self) -> Result<PathBuf, Error> {
            let dir = tempfile::tempdir().expect("tempdir");
            let path = dir.keep().join("flock.json");
            if self.unreadable_roll {
                // A field of the wrong type, which is what `AppConfig`
                // still refuses. An unknown field was the trigger until
                // shep-core 0.7 dropped `deny_unknown_fields`, at which
                // point this roll became readable and the test below
                // asserted nothing. Either way the point is the same: the
                // roll fails on its own crafted, actionable cause rather
                // than on one this double invented.
                fs::write(&path, "{\"apps\":[{\"app\":{\"name\":123}}]}").expect("write roll");
                return Ok(path);
            }
            let apps: Vec<String> = self
                .sheep
                .iter()
                .map(|name| {
                    let cwd = self
                        .registered_cwd
                        .clone()
                        .unwrap_or_else(|| "/srv/deploy-tree/current".to_owned());
                    format!(
                        "{{\"app\":{{\"name\":{name:?},\"script\":\"the-shepherds-own-script\",\
                         \"cwd\":{cwd:?}}}}}"
                    )
                })
                .collect();
            fs::write(&path, format!("{{\"apps\":[{}]}}", apps.join(","))).expect("write roll");
            Ok(path)
        }

        crate::fixtures::daemon_methods!(unimplemented;
            dog_config, list_flock, reload, restart, set_smit,
        );
    }

    /// fails if a sheep that pre-existed the dog is not put back where its
    /// operator will look for it. This is the whole point: they will go to
    /// ~/ReactMap, because that is where they think their app lives, and
    /// the cwd it has been running under is a path beneath $SHEP_HOME they
    /// have no reason to know about.
    #[tokio::test]
    async fn a_pre_existing_sheep_goes_back_to_its_own_checkout() {
        let home = tempfile::tempdir().expect("tempdir");
        write_target_with_origin(home.path(), "bpm", "/srv/reactmap", "bun .");
        let daemon = Recording::with_registered(&["bpm"]);

        let results = all(&daemon, home.path()).await;

        assert!(matches!(results[0], Restored::Returned { .. }));
        let started = daemon.started();
        assert_eq!(started[0].cwd.as_deref(), Some("/srv/reactmap"));
        assert_eq!(started[0].script, "bun .");
    }

    /// fails if the restore stops deleting the old registration first. The
    /// registered config is what has to change, and `Start` on a registered
    /// name ADDS an instance rather than re-registering it, so without the
    /// delete the sheep ends up running from both places at once.
    #[tokio::test]
    async fn the_old_registration_is_removed_before_the_new_one_is_started() {
        let home = tempfile::tempdir().expect("tempdir");
        write_target_with_origin(home.path(), "bpm", "/srv/reactmap", "bun .");
        let daemon = Recording::with_registered(&["bpm"]);

        all(&daemon, home.path()).await;

        assert_eq!(
            daemon.calls(),
            vec!["delete", "start"],
            "deleting after starting would leave two registrations"
        );
    }

    /// fails if a sheep the dog bootstrapped is deleted, or is left without
    /// being told about. Deleting an app because a deploy tool was
    /// uninstalled would be much worse than leaving it, and "left running,
    /// unchanged" that nobody is told about is indistinguishable from
    /// "quietly abandoned somewhere you will not think to look".
    #[tokio::test]
    async fn a_bootstrapped_sheep_is_left_running_and_named_in_the_report() {
        let home = tempfile::tempdir().expect("tempdir");
        write_target_with_origin_absent(home.path(), "ctm");
        let daemon = Recording::with_registered(&["ctm"]);

        let results = all(&daemon, home.path()).await;

        assert!(daemon.calls().is_empty(), "nothing is stopped or started");
        let text = report(&results);
        assert!(text.contains("ctm still running from"), "{text}");
        assert!(text.contains("deploy/ctm/current"), "{text}");
    }

    /// fails if a tree whose cutover never landed is "restored".
    ///
    /// `optin::prepare` writes `origin_cwd` and `origin_script` and then
    /// stops. `cut_over` is what actually moves the sheep. So between them a
    /// tree exists naming an origin for a sheep that never left its own
    /// checkout, and the branch here read those two fields alone: it could
    /// not tell that tree from a sheep genuinely cut over.
    ///
    /// Restoring it deletes and restarts a healthy app to put it back where
    /// it already is. Worse if the `Start` then fails, because the delete has
    /// already landed and the app is simply stopped. `deployed` is the field
    /// that says a cutover landed.
    #[tokio::test]
    async fn a_cutover_that_never_landed_is_not_restored() {
        let home = tempfile::tempdir().expect("tempdir");
        write_target_never_cut_over(home.path(), "ctm", "/srv/ctm", "./run.sh");
        let daemon = Recording::with_registered(&["ctm"]);

        let results = all(&daemon, home.path()).await;

        assert!(
            daemon.calls().is_empty(),
            "a sheep that never moved must not be stopped or started: {:?}",
            daemon.calls()
        );
        let text = report(&results);
        assert!(text.contains("ctm was never moved"), "{text}");
        assert!(text.contains("/srv/ctm"), "{text}");
    }

    /// fails if a cutover that got as far as registering is called "never
    /// moved" and left alone.
    ///
    /// `cut_over` writes `deployed` only on its fully verified path. Its
    /// `NotVerified` and `Failed` arms return without touching it, having
    /// already run `undo_start`, and `undo_start`'s own repair can fail and
    /// leave a newcomer registered. So an absent `deployed` means the cutover
    /// did not LAND, not that it never RAN, and reading it alone reported a
    /// sheep in that state as running where it belongs, doing nothing, at the
    /// one moment this module exists to catch exactly that.
    ///
    /// The shepherd's own registration is what separates the two, because
    /// `cut_over` is the only thing that ever points a sheep's `cwd` at the
    /// tree's `current`.
    #[tokio::test]
    async fn a_cutover_that_registered_is_restored_even_without_a_deployed_sha() {
        let home = tempfile::tempdir().expect("tempdir");
        write_target_never_cut_over(home.path(), "ctm", "/srv/ctm", "./run.sh");
        let tree = Tree::for_sheep(home.path(), "ctm");
        let daemon = Recording::with_registered(&["ctm"]).registered_at(&tree.current());

        let results = all(&daemon, home.path()).await;

        let text = report(&results);
        assert!(
            !text.contains("never moved"),
            "a sheep the cutover registered against the tree was moved: {text}"
        );
        assert!(
            !daemon.calls().is_empty(),
            "it must actually be put back, not merely reported"
        );
    }

    /// Writes a `deploy.toml` as `optin::prepare` leaves one when its cutover
    /// never ran: an origin recorded, and no `deployed`.
    fn write_target_never_cut_over(home: &Path, sheep: &str, origin_cwd: &str, script: &str) {
        let tree = Tree::for_sheep(home, sheep);
        fs::create_dir_all(tree.state_file().parent().expect("has a parent"))
            .expect("create target dir");
        let state = State {
            origin_cwd: Some(PathBuf::from(origin_cwd)),
            origin_script: Some(script.to_owned()),
            checkout: PathBuf::from(origin_cwd),
            ..fixtures::state()
        };
        state.write(&tree.state_file()).expect("write state");
    }

    /// fails if one target's failure stops the others being restored, or
    /// stops the removal. An operator asking to remove something is
    /// entitled to have it removed, and a dog that refused to be
    /// uninstalled because one of five sheep would not restart would be
    /// worse than one that did nothing at all.
    #[tokio::test]
    async fn a_failure_is_reported_and_the_rest_still_run() {
        let home = tempfile::tempdir().expect("tempdir");
        write_target_with_origin(home.path(), "aaa", "/srv/a", "./a");
        write_target_with_origin(home.path(), "zzz", "/srv/z", "./z");
        let daemon = Recording::refusing_first_start_only(&["aaa", "zzz"]);

        let results = all(&daemon, home.path()).await;

        assert_eq!(results.len(), 2);
        // `Reset`, not `Lost`: the fallback re-registered what the shepherd
        // already had, so "aaa" is still running. A double that refused
        // EVERY start would give `Lost` here, which is a different claim and
        // has its own test below.
        assert!(
            matches!(results[0], Restored::Reset { .. }),
            "{:?}",
            results[0]
        );
        assert!(matches!(results[1], Restored::Returned { .. }));
        assert!(report(&results).contains("aaa"), "the failure is named");
    }

    /// fails if a refused restore leaves the sheep deleted when it did not
    /// have to be. `Delete` is stop plus deregister and the roll drops a
    /// name with no live instance, so the window between the delete and a
    /// refused `Start` is one where the sheep is gone from both. The
    /// fallback re-registers what the shepherd had a moment ago, which is
    /// the right answer for the common causes: a transient refusal, a bad
    /// origin_script, a `user` that no longer resolves.
    #[tokio::test]
    async fn a_refused_restore_puts_the_shepherds_own_config_back() {
        let home = tempfile::tempdir().expect("tempdir");
        write_target_with_origin(home.path(), "bpm", "/srv/reactmap", "bun .");
        let daemon = Recording::refusing_first_start_only(&["bpm"]);

        let results = all(&daemon, home.path()).await;

        assert!(
            matches!(results[0], Restored::Reset { .. }),
            "{:?}",
            results[0]
        );
        assert_eq!(daemon.started().len(), 2, "the restore, then the fallback");
        assert_eq!(
            daemon.started()[1].cwd.as_deref(),
            Some("/srv/deploy-tree/current"),
            "the fallback re-registers what the shepherd had, not the restore"
        );
    }

    /// fails if a rescued restore is reported with wording that claims
    /// nothing happened. The delete already ran and `start(vec![current])`
    /// then succeeded, so the sheep was stopped and a fresh instance
    /// started - not "left as it is" in any sense an operator would
    /// recognise, even though the registered config now matches what it
    /// was a moment ago.
    #[tokio::test]
    async fn a_rescued_restore_is_worded_as_a_reset_not_left_alone() {
        let home = tempfile::tempdir().expect("tempdir");
        write_target_with_origin(home.path(), "bpm", "/srv/reactmap", "bun .");
        let daemon = Recording::refusing_first_start_only(&["bpm"]);

        let results = all(&daemon, home.path()).await;

        let text = report(&results);
        assert!(!text.contains("left as it is"), "{text}");
        assert!(text.contains("stopped it and started it again"), "{text}");
    }

    /// fails if a sheep that really has been deleted is reported as merely
    /// "could not be restored". Every other outcome here leaves a running
    /// app; this one does not, and an operator reading the gentler wording
    /// would assume theirs was still up. This is the one row in the whole
    /// report that has to be alarming.
    #[tokio::test]
    async fn a_sheep_left_deleted_says_so_in_those_words() {
        let home = tempfile::tempdir().expect("tempdir");
        write_target_with_origin(home.path(), "bpm", "/srv/reactmap", "bun .");
        let daemon = Recording::refusing_every_start(&["bpm"]);

        let results = all(&daemon, home.path()).await;

        assert!(
            matches!(results[0], Restored::Lost { .. }),
            "{:?}",
            results[0]
        );
        let text = report(&results);
        assert!(text.contains("NO LONGER REGISTERED"), "{text}");
        assert!(text.contains("will not come back"), "{text}");
    }

    /// fails if the deploy tree is removed. It is not the dog's to delete,
    /// and in the bootstrap case a running app is still pointing into it,
    /// so deleting it would take down an app during an uninstall.
    #[tokio::test]
    async fn the_deploy_tree_is_left_on_disk() {
        let home = tempfile::tempdir().expect("tempdir");
        write_target_with_origin(home.path(), "bpm", "/srv/reactmap", "bun .");
        all(&Recording::with_registered(&["bpm"]), home.path()).await;
        assert!(home.path().join("deploy/bpm/deploy.toml").is_file());
    }

    /// fails if a muster roll that cannot be read produces a separate,
    /// fabricated "not registered" row per pre-existing sheep instead of
    /// one row naming all of them and the roll's own real cause. An
    /// operator meeting N copies of a wrong guess has less to act on than
    /// one line naming the actual reason - here, `roll::read`'s "newer
    /// shepherd" message - and the affected sheep.
    #[tokio::test]
    async fn a_roll_read_failure_is_reported_once_dog_wide_with_the_real_cause() {
        let home = tempfile::tempdir().expect("tempdir");
        write_target_with_origin(home.path(), "aaa", "/srv/a", "./a");
        write_target_with_origin(home.path(), "zzz", "/srv/z", "./z");
        let daemon = Recording::with_unreadable_roll();

        let results = all(&daemon, home.path()).await;

        assert_eq!(
            results.len(),
            1,
            "one row for the whole roll failure, not one per sheep: {results:?}"
        );
        assert!(matches!(results[0], Restored::RollUnreadable { .. }));
        let text = report(&results);
        assert!(text.contains("aaa"), "{text}");
        assert!(text.contains("zzz"), "{text}");
        assert!(text.contains("newer"), "{text}");
        assert!(
            daemon.calls().is_empty(),
            "nothing was deleted or started without a readable roll"
        );
    }

    /// fails if a shepherd that refuses `describe` outright is reported as
    /// anything other than `Failed`, or if the refusal is silently
    /// swallowed into `Done`. `put_back` cannot learn what to delete
    /// without this call, so nothing about the sheep changes.
    #[tokio::test]
    async fn a_describe_that_is_refused_leaves_the_sheep_untouched() {
        let home = tempfile::tempdir().expect("tempdir");
        write_target_with_origin(home.path(), "bpm", "/srv/reactmap", "bun .");
        let daemon = Recording::refusing_describe(&["bpm"]);

        let results = all(&daemon, home.path()).await;

        assert!(
            matches!(results[0], Restored::Failed { .. }),
            "{:?}",
            results[0]
        );
        assert!(daemon.calls().is_empty(), "nothing was deleted or started");
    }

    /// fails if a `delete` that fails PARTWAY through a multi-instance
    /// sheep is reported as anything other than its own `PartlyDeleted`, or
    /// if the loop keeps deleting instances after one refuses. This is the
    /// single most dangerous path in the file: some instances may already
    /// be gone and others still live, and neither `Failed`'s "left as it
    /// is" nor `Lost`'s "gone from the flock" is a true sentence about it -
    /// only `shep describe` can say what is really there.
    #[tokio::test]
    async fn a_delete_that_fails_partway_through_is_reported_as_partly_deleted() {
        let home = tempfile::tempdir().expect("tempdir");
        write_target_with_origin(home.path(), "bpm", "/srv/reactmap", "bun .");
        let daemon = Recording::refusing_delete_partway(&["bpm"]);

        let results = all(&daemon, home.path()).await;

        assert!(
            matches!(results[0], Restored::PartlyDeleted { .. }),
            "{:?}",
            results[0]
        );
        assert_eq!(
            daemon.calls(),
            vec!["delete"],
            "the first delete lands before the second refuses, and nothing after it runs"
        );
        assert!(
            daemon.started().is_empty(),
            "start is never reached once a delete fails"
        );
    }

    /// fails if a partly-deleted sheep is worded as though it were either
    /// fully running or fully removed. Neither `Failed`'s "left as it is"
    /// nor `Lost`'s "gone from the flock" is true here, and this pins the
    /// two true things the report can say: check `shep describe` for what
    /// is actually there, and `shep flock` for whether it is still
    /// registered - this crate cannot know that from here, since `Delete`
    /// landing on the LAST remaining instance really does deregister it.
    #[tokio::test]
    async fn a_partly_deleted_sheep_is_worded_as_neither_running_nor_removed() {
        let home = tempfile::tempdir().expect("tempdir");
        write_target_with_origin(home.path(), "bpm", "/srv/reactmap", "bun .");
        let daemon = Recording::refusing_delete_partway(&["bpm"]);

        let results = all(&daemon, home.path()).await;

        let text = report(&results);
        assert!(!text.contains("left as it is"), "{text}");
        assert!(!text.contains("gone from the flock"), "{text}");
        assert!(!text.contains("never touched"), "{text}");
        assert!(
            text.contains("neither fully running nor fully removed"),
            "{text}"
        );
        assert!(text.contains("shep describe bpm"), "{text}");
        assert!(text.contains("shep flock"), "{text}");
    }

    /// fails if a delete refused on the FIRST instance, with nothing behind
    /// it, is reported as `PartlyDeleted`. Nothing landed: the sheep is
    /// fully running and fully registered, which is exactly `Failed`'s
    /// "left as it is" case. The discriminator for `PartlyDeleted` is
    /// whether a delete already landed, not whether one failed - a
    /// single-instance sheep whose only delete is refused must stay
    /// `Failed`, and this is the ordinary shape a refused delete takes.
    #[tokio::test]
    async fn a_delete_refused_with_nothing_landed_yet_stays_failed() {
        let home = tempfile::tempdir().expect("tempdir");
        write_target_with_origin(home.path(), "bpm", "/srv/reactmap", "bun .");
        let daemon = Recording::refusing_first_delete_only(&["bpm"]);

        let results = all(&daemon, home.path()).await;

        assert!(
            matches!(results[0], Restored::Failed { .. }),
            "{:?}",
            results[0]
        );
        let text = report(&results);
        assert!(text.contains("nothing was changed by this dog"), "{text}");
        assert!(
            !text.contains("neither fully running nor fully removed"),
            "{text}"
        );
    }

    /// fails if `on-remove` ever turns a partial failure into a nonzero
    /// exit. There is no such branch in `main.rs`'s `on_remove` - this pins
    /// the half of that contract that lives here: a report containing a
    /// `Failed` row still names every other row plainly, so nothing about
    /// this module's own output would justify one.
    #[test]
    fn a_report_with_a_failure_still_names_every_other_row() {
        let results = vec![
            Restored::Failed {
                sheep: "aaa".to_owned(),
                why: "refused".to_owned(),
            },
            Restored::Returned {
                sheep: "zzz".to_owned(),
                to: PathBuf::from("/srv/z"),
            },
        ];
        let text = report(&results);
        assert!(text.contains("aaa"), "{text}");
        assert!(text.contains("zzz"), "{text}");
    }

    /// fails if a deploy directory that exists but cannot be listed is
    /// swallowed into an empty report. `paths::targets` returns that error
    /// rather than treating it as "nothing here", and dropping it left
    /// `report(&[])` printing nothing and `on_remove` exiting 0 with every
    /// sheep under the tree left running.
    #[tokio::test]
    async fn a_deploy_directory_that_cannot_be_listed_is_reported_not_silenced() {
        let home = tempfile::tempdir().expect("tempdir");
        // A file where a directory is expected makes `read_dir` fail with
        // something other than `NotFound`, which `paths::targets` does not
        // treat as "nothing to restore".
        fs::write(home.path().join("deploy"), b"not a directory").expect("write file");

        let results = all(&Recording::new(&[], Refuse::Never), home.path()).await;

        assert_eq!(
            results.len(),
            1,
            "the failure must be reported, not swallowed: {results:?}"
        );
        assert!(
            matches!(&results[0], Restored::Unlisted { dir, .. } if dir.ends_with("deploy")),
            "got: {:?}",
            results[0]
        );
        assert!(
            !report(&results).is_empty(),
            "silence here reads as success"
        );
    }

    /// fails if a run with no deploy targets at all prints nothing. Silence
    /// is indistinguishable from success, and `main.rs`'s sibling branch
    /// prints a sentence for exactly this reason.
    #[test]
    fn an_empty_report_says_there_was_nothing_to_restore() {
        assert_eq!(report(&[]), "no deploy targets, nothing to restore\n");
    }

    /// fails if the dog and the daemon spell `$SHEP_HOME` differently and a
    /// sheep the cutover DID register reads as never moved because of it.
    /// `daemon.rs:141` says the two can resolve the same directory through
    /// different paths, so the comparison has to see through that rather
    /// than compare spellings literally.
    #[tokio::test]
    async fn a_registration_reached_through_a_symlinked_parent_is_recognised() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let home = tmp.path().join("home");
        fs::create_dir_all(&home).expect("create home");
        let link = tmp.path().join("link");
        std::os::unix::fs::symlink(&home, &link).expect("symlink parent");

        let tree = Tree::for_sheep(&home, "ctm");
        fs::create_dir_all(tree.root()).expect("create tree");
        let release = tmp.path().join("release");
        fs::create_dir_all(&release).expect("create release dir");
        std::os::unix::fs::symlink(&release, tree.current()).expect("symlink current");

        write_target_never_cut_over(&home, "ctm", "/srv/ctm", "./run.sh");
        let cwd_through_link = link.join("deploy").join("ctm").join("current");
        let daemon = Recording::with_registered(&["ctm"]).registered_at(&cwd_through_link);

        let results = all(&daemon, &home).await;

        let text = report(&results);
        assert!(
            !text.contains("never moved"),
            "the two spellings resolve to the same directory: {text}"
        );
        assert!(
            !daemon.calls().is_empty(),
            "it must actually be put back, not merely reported"
        );
    }

    /// fails if the same recognition breaks once the release `current`
    /// pointed at is gone. Both full canonicalisations fail on a dangling
    /// link, and a fallback to the literal spellings called the two names
    /// for one `current` different, so the sheep read as never moved and
    /// stayed registered against the tree.
    #[tokio::test]
    async fn a_registration_through_a_symlinked_parent_is_recognised_when_current_dangles() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let home = tmp.path().join("home");
        fs::create_dir_all(&home).expect("create home");
        let link = tmp.path().join("link");
        std::os::unix::fs::symlink(&home, &link).expect("symlink parent");

        let tree = Tree::for_sheep(&home, "ctm");
        fs::create_dir_all(tree.root()).expect("create tree");
        std::os::unix::fs::symlink(tree.release("gone"), tree.current()).expect("dangling current");

        write_target_never_cut_over(&home, "ctm", "/srv/ctm", "./run.sh");
        let cwd_through_link = link.join("deploy").join("ctm").join("current");
        let daemon = Recording::with_registered(&["ctm"]).registered_at(&cwd_through_link);

        let results = all(&daemon, &home).await;

        let text = report(&results);
        assert!(!text.contains("never moved"), "{text}");
        assert!(!daemon.calls().is_empty(), "put back, not merely reported");
    }
}