repon 0.30.5

//! The three built-in management operations: `ignore`, `delete` and `sync`.
//!
//! [repo-management.md](../../../docs/spec/repo-management.md) is the specification and
//! [0028](../../../docs/adr/0028-repon-writes-the-repo-entries-it-owns.md) the reasoning for
//! the first three. They are built-in entries in the Action palette
//! rather than a third palette, and they fan out over the Selection sharing the Action
//! confirm gate's shape (a count, with ineligible entities subtracted and named) and none of
//! the pty machinery in [actions.md](../../../docs/spec/actions.md) for the operation itself,
//! because no child process runs for it. `sync` fans out no mutation of its own either: it
//! reuses [`repon_core::Core::attempt_auto_update`], the identical fast-forward the periodic
//! fetch's own auto-update already runs. `sync` alone may still spawn a child process, through
//! the `before_sync` and `after_sync` hooks a Set may declare around it
//! ([repo-management.md](../../../docs/spec/repo-management.md)'s "Hooks around sync",
//! [0032](../../../docs/adr/0032-hooks-around-a-built-in-fire-on-its-own-confirm-gate-never-its-completion.md)),
//! run through [`repon_core::Core::run_action_for_entity_blocking`] rather than the async
//! fan-out `on_refresh` and the palette share.
//!
//! What a run leaves behind, per repo-management.md's "Receipts": an
//! [`repon_core::ActionReceipt`] whose single Step is the act Repon performed itself, carrying
//! [`repon_core::OwnWork`] rather than an exit code, since no child process ran. [`Outcome`]
//! is this module's own vocabulary for what happened to one row and [`own_work`] is the one
//! place it turns into the receipt's words, which the log line reads too so the two cannot
//! drift. The receipt does not replace the confirm gate: that still names and counts every
//! refusal before the gesture is accepted, which is where repo-management.md's own "What
//! `delete` refuses" puts it.

use std::{
    fs,
    path::{Path, PathBuf},
    sync::Arc,
    time::{Duration, Instant},
};

use color_eyre::eyre::{Result, eyre};
use repon_core::{AutoUpdateAttempt, DeleteRisk, EntityKey, EntityState, Kind, OwnWork};

use crate::config::repo_entry::{self, Edit};
use crate::selection::{RunScope, Targets};

/// One of the three built-in entries in the Action palette, in the order
/// [repo-management.md](../../../docs/spec/repo-management.md)'s own operations table lists
/// them.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Operation {
    Ignore,
    Delete,
    Sync,
}

/// Every built-in operation, which is also the list `m` filters the palette down to and the
/// set of names a config-defined `[[action]]` may not take
/// ([`crate::config::document`]'s own load-time check reads this).
pub(crate) const OPERATIONS: [Operation; 3] =
    [Operation::Ignore, Operation::Delete, Operation::Sync];

impl Operation {
    /// The name the palette lists it under, and the reserved name a config-defined
    /// `[[action]]` may not take.
    pub(crate) fn name(self) -> &'static str {
        match self {
            Operation::Ignore => "ignore",
            Operation::Delete => "delete",
            Operation::Sync => "sync",
        }
    }

    /// The palette's own second column, in the same slot a config-defined Action's
    /// `description` occupies.
    pub(crate) fn description(self) -> &'static str {
        match self {
            Operation::Ignore => "Hide the selected entities, or show them again",
            Operation::Delete => "Remove the selected working trees, permanently",
            Operation::Sync => "Fast-forward the selected Repos to their tracked upstream",
        }
    }

    /// The name whose reserved status is what a config-defined `[[action]]` collides with.
    pub(crate) fn from_name(name: &str) -> Option<Operation> {
        OPERATIONS
            .into_iter()
            .find(|operation| operation.name() == name)
    }

    /// Whether this operation widens to every visible row on an empty Selection, the way a
    /// declared Action already does, rather than falling back to the cursor row alone.
    /// `sync` alone widens; `ignore` and, safety-critically, `delete` (which
    /// permanently removes working trees) keep the cursor-row fallback
    /// ([actions.md](../../../docs/spec/actions.md)'s "The Selection and the gate").
    pub(crate) fn widens_to_every_visible_row_when_selection_is_empty(self) -> bool {
        match self {
            Operation::Sync => true,
            Operation::Ignore | Operation::Delete => false,
        }
    }

    /// Whether `entity` is operated on, or the reason it is not, per
    /// [repo-management.md](../../../docs/spec/repo-management.md)'s "eligible" column. Every
    /// pairing of the four operations with the three Kinds is named here rather than falling
    /// through a catch-all, so a fifth Kind fails to compile instead of quietly becoming
    /// eligible for a destructive operation.
    ///
    /// What the auto-update's own five rules find ineligible right now (dirty, no upstream,
    /// not behind, not fast-forward) is a different fact, read only by attempting it, so it
    /// is never a gate refusal here; [`run`]'s own `sync_one` is where that surfaces.
    pub(crate) fn eligibility(self, entity: &EntityState) -> Eligibility {
        match (self, entity.kind) {
            (Operation::Ignore, Kind::Repo | Kind::Worktree) => Eligibility::Eligible,
            (Operation::Ignore, Kind::Submodule) => {
                Eligibility::Refused(Refusal::SubmoduleHasNoEntryOfItsOwn)
            }
            (Operation::Delete, Kind::Repo | Kind::Worktree) => Eligibility::Eligible,
            (Operation::Delete, Kind::Submodule) => {
                Eligibility::Refused(Refusal::SubmoduleCannotBeDeleted)
            }
            (Operation::Sync, Kind::Repo) => Eligibility::Eligible,
            (Operation::Sync, Kind::Worktree) => {
                Eligibility::Refused(Refusal::WorktreeSyncsThroughItsRepo)
            }
            (Operation::Sync, Kind::Submodule) => {
                Eligibility::Refused(Refusal::SubmoduleCannotSync)
            }
        }
    }
}

/// Whether a Selection row is operated on, or the reason it is not. The refused half is
/// reported and counted in the confirm gate rather than dropped, the same way an excluded
/// entity is subtracted and named ([actions.md](../../../docs/spec/actions.md)).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Eligibility {
    Eligible,
    Refused(Refusal),
}

/// Why one Selection row is not operated on.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Refusal {
    /// `delete` on a Submodule: its git common dir is `<parent>/.git/modules/<name>` rather
    /// than its own, so removing the directory corrupts the parent, whose `.gitmodules`
    /// still names it.
    SubmoduleCannotBeDeleted,
    /// `ignore` on a Submodule: a `[[repo]]` entry's `path` resolves to a git common dir,
    /// and a Submodule's is its parent's `.git/modules/<name>`, so one entry cannot cover a
    /// parent and its Submodules together
    /// ([config.md](../../../docs/spec/config.md)'s per-Repo entries).
    SubmoduleHasNoEntryOfItsOwn,
    /// `sync` on a Worktree: the auto-update it reuses acts on a Repo's own branch, and
    /// `repon-core`'s own `repos_eligible_for_auto_update_attempt` is Repo-only for exactly
    /// that reason, so a Worktree sharing a common dir with a Repo is refused rather than
    /// silently doing nothing.
    WorktreeSyncsThroughItsRepo,
    /// `sync` on a Submodule: it tracks a pinned commit, not a branch, so there is nothing
    /// to fast-forward.
    SubmoduleCannotSync,
}

impl Refusal {
    /// The reason the confirm gate shows beside the entity's name.
    pub(crate) fn reason(self) -> &'static str {
        match self {
            Refusal::SubmoduleCannotBeDeleted => {
                "a Submodule's git dir lives in its parent; deleting it corrupts the parent"
            }
            Refusal::SubmoduleHasNoEntryOfItsOwn => {
                "a Submodule shares its parent's `[[repo]]` entry and has none of its own"
            }
            Refusal::WorktreeSyncsThroughItsRepo => {
                "sync acts on a Repo's own branch; a Worktree shares it and is not itself \
                 the target"
            }
            Refusal::SubmoduleCannotSync => {
                "a Submodule tracks a pinned commit, not a branch, so there is nothing to \
                 fast-forward"
            }
        }
    }
}

/// One Selection row as the confirm gate sees it: what it is, whether the operation will act
/// on it, and, for a `delete` that will, what accepting destroys.
#[derive(Debug, Clone)]
pub(crate) struct Target {
    pub(crate) key: EntityKey,
    pub(crate) name: Arc<str>,
    pub(crate) kind: Kind,
    /// Shared with every other Entity attached to the same Repo, which is what
    /// [`drop_worktrees_covered_by_their_own_selected_parent`] matches a Worktree against
    /// its parent Repo by, rather than by path.
    pub(crate) common_dir: Arc<Path>,
    pub(crate) eligibility: Eligibility,
    /// Whether a `[[repo]]` entry already excludes this row, which is what `ignore` reads
    /// to decide which way to run ([`run_one`]).
    pub(crate) excluded: bool,
    /// `delete` only, and only on a row it will act on: `Ok` with the read, or `Err` with
    /// why it could not be read, never a zeroed stand-in.
    pub(crate) risk: Option<Result<DeleteRisk, String>>,
}

/// The whole gesture, resolved before anything acts: which operation, and every Selection row
/// with its verdict. Built once when the gate opens, so the count the gate shows and the rows
/// the run acts on cannot disagree.
#[derive(Debug, Clone)]
pub(crate) struct Plan {
    pub(crate) operation: Operation,
    pub(crate) targets: Vec<Target>,
    /// Which rows [`Plan::new`] was handed, carried through so the gate's headline names the
    /// same scope the palette's border title above it does.
    pub(crate) scope: RunScope,
}

impl Plan {
    /// `operation` resolved against the Selection: every key in `targets`, in the Selection's
    /// own order, paired with the verdict [`Operation::eligibility`] gives it. A key the
    /// snapshot no longer holds is dropped, the same fallback every key-addressed entry point
    /// on [`repon_core::Core`] gives one.
    ///
    /// Cheap: it reads the snapshot and nothing else, so the palette's border count can be
    /// rebuilt every frame. [`Plan::with_risk`] is the expensive half, run once when the gate
    /// opens.
    pub(crate) fn new(operation: Operation, entities: &[EntityState], targets: Targets) -> Self {
        let mut plan_targets: Vec<Target> = targets
            .keys
            .iter()
            .filter_map(|key| entities.iter().find(|entity| &entity.key == key))
            .map(|entity| Target {
                key: entity.key.clone(),
                name: Arc::clone(&entity.name),
                kind: entity.kind,
                common_dir: Arc::clone(&entity.common_dir),
                eligibility: operation.eligibility(entity),
                excluded: entity.excluded,
                risk: None,
            })
            .collect();
        if operation == Operation::Delete {
            drop_worktrees_covered_by_their_own_selected_parent(&mut plan_targets);
        }
        Plan {
            operation,
            targets: plan_targets,
            scope: targets.scope,
        }
    }

    /// Reads what accepting destroys, once, for every row a `delete` will act on. `read` is
    /// [`repon_core::Core::delete_risk`] at the one call site; taken as a parameter so this
    /// module never needs a `Core` to be tested. A no-op for `ignore`, which
    /// destroy nothing and so get the ordinary gate with no additional lines.
    pub(crate) fn with_risk(
        mut self,
        read: impl Fn(&EntityKey) -> std::result::Result<DeleteRisk, String>,
    ) -> Self {
        if self.operation != Operation::Delete {
            return self;
        }
        for target in &mut self.targets {
            if target.eligibility == Eligibility::Eligible {
                target.risk = Some(read(&target.key));
            }
        }
        self
    }

    /// How many rows the run will act on: the Selection with this operation's own ineligible
    /// rows subtracted, which is the number the palette's border and the gate's headline both
    /// read.
    pub(crate) fn eligible_count(&self) -> usize {
        self.targets
            .iter()
            .filter(|target| target.eligibility == Eligibility::Eligible)
            .count()
    }

    /// How many were named and subtracted rather than dropped
    /// ([repo-management.md](../../../docs/spec/repo-management.md): a refusal is "reported
    /// and counted in the confirm gate, never silent").
    pub(crate) fn refused_count(&self) -> usize {
        self.targets.len() - self.eligible_count()
    }

    /// The gate's own lines: the headline count with the refusals subtracted and counted,
    /// then one line per row, then the sentence saying in as many words that there is no undo
    /// and no trash. `ignore` gets the ordinary gate with no additional lines,
    /// since neither destroys anything
    /// ([repo-management.md](../../../docs/spec/repo-management.md)'s "The confirm gate").
    pub(crate) fn confirm_lines(&self) -> Vec<String> {
        let mut lines = vec![headline(
            self.operation,
            self.scope,
            self.eligible_count(),
            self.refused_count(),
        )];
        for target in &self.targets {
            lines.push(target_line(self.operation, target));
        }
        if self.operation == Operation::Delete {
            lines.push(NO_UNDO.to_string());
        }
        lines
    }
}

/// Drops a Worktree target whose parent Repo is also targeted, so a `delete` over both
/// reports one removal rather than two: the Repo's own run already takes its linked
/// Worktrees with it
/// ([repo-management.md](../../../docs/spec/repo-management.md)'s "Deleting a Repo also
/// takes its linked Worktrees with it"). Matched by `common_dir` rather than by path, the
/// same fact that ties a Worktree to the Repo it shares an object store with.
fn drop_worktrees_covered_by_their_own_selected_parent(targets: &mut Vec<Target>) {
    let selected_repos: std::collections::HashSet<Arc<Path>> = targets
        .iter()
        .filter(|target| target.kind == Kind::Repo)
        .map(|target| Arc::clone(&target.common_dir))
        .collect();
    targets.retain(|target| {
        target.kind != Kind::Worktree || !selected_repos.contains(&target.common_dir)
    });
}

/// The gate's own sentence about permanence
/// ([repo-management.md](../../../docs/spec/repo-management.md): "There is no undo and no
/// trash, which the gate says in as many words").
pub(crate) const NO_UNDO: &str = "there is no undo and no trash";

/// The Notice `App::run_management` raises the instant the confirm gate is accepted, replacing
/// the gate itself before any of the operation's blocking work starts: [`Report::summary`]
/// replaces it once that work finishes. Named after `eligible_count`, the same row count the
/// gate's own [`headline`] showed, since the gate and the run must agree on how many rows this
/// is about, and on which rows they were.
pub(crate) fn running_notice(operation: Operation, scope: RunScope, eligible: usize) -> String {
    format!(
        "{}: running on {eligible} {}",
        operation.name(),
        scope.word()
    )
}

/// The Notice `App::run_management` paints before each row's own work starts, replacing
/// [`running_notice`] as the run goes: the operation, the row's own name, and its position
/// among every row this run visits in order. A refused row is still visited and still
/// counts towards `total`, since the run reports it too rather than skipping straight past
/// it; only an eligible row's own work can stall, but naming a refused row here as it is
/// reached keeps the position honest over the whole Selection the gate showed.
pub(crate) fn row_notice(
    operation: Operation,
    name: &str,
    position: usize,
    total: usize,
) -> String {
    format!("{}: {name} ({position}/{total})", operation.name())
}

fn headline(operation: Operation, scope: RunScope, eligible: usize, refused: usize) -> String {
    let name = operation.name();
    let scope = scope.word();
    if refused == 0 {
        format!("{name} on {eligible} {scope}?")
    } else {
        let total = eligible + refused;
        format!("{name} on {eligible} of {total} {scope}, {refused} refused?")
    }
}

/// One row's line in the gate: its name plus its refusal reason, or, for a `delete` it will
/// act on, the risk lines repo-management.md's "The confirm gate" names. A Repo with none of
/// the three is listed plainly, which is its name and nothing else.
fn target_line(operation: Operation, target: &Target) -> String {
    match target.eligibility {
        Eligibility::Refused(refusal) => {
            format!("{}: refused, {}", target.name, refusal.reason())
        }
        Eligibility::Eligible => match (operation, &target.risk) {
            (Operation::Delete, Some(Ok(risk))) => match risk_phrases(risk, target.kind) {
                phrases if phrases.is_empty() => target.name.to_string(),
                phrases => format!("{}: {}", target.name, phrases.join(", ")),
            },
            (Operation::Delete, Some(Err(error))) => {
                format!(
                    "{}: what it would destroy could not be read, {error}",
                    target.name
                )
            }
            (Operation::Delete, None) | (Operation::Ignore | Operation::Sync, _) => {
                target.name.to_string()
            }
        },
    }
}

/// The facts the gate names per row, each present only when it is true, so a row with none
/// of them produces an empty list and is listed plainly. A Repo's linked-Worktree count
/// names what its own `delete` destroys along with it; a Worktree row never carries that
/// phrase, because deleting one Worktree never touches its siblings.
fn risk_phrases(risk: &DeleteRisk, kind: Kind) -> Vec<String> {
    let DeleteRisk {
        uncommitted,
        unpushed_commits,
        unpushed_branches,
        linked_worktrees,
    } = *risk;
    let mut phrases = Vec::new();
    if uncommitted {
        phrases.push("uncommitted changes".to_string());
    }
    if unpushed_commits > 0 {
        phrases.push(format!(
            "{unpushed_commits} {} unpushed on {unpushed_branches} {}",
            plural(unpushed_commits, "commit", "commits"),
            plural(unpushed_branches, "branch", "branches"),
        ));
    }
    if kind == Kind::Repo && linked_worktrees > 0 {
        phrases.push(format!(
            "{linked_worktrees} linked {}",
            plural(linked_worktrees, "worktree", "worktrees")
        ));
    }
    phrases
}

fn plural(count: u32, one: &'static str, many: &'static str) -> &'static str {
    if count == 1 { one } else { many }
}

/// Which of `delete`'s three removals one row confirmed, which is its receipt's own first
/// clause ([repo-management.md](../../../docs/spec/repo-management.md)'s "Receipts").
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Removal {
    /// A Repo's working tree, along with every linked Worktree's own directory
    /// ([repo-management.md](../../../docs/spec/repo-management.md)'s "What `delete` does to
    /// a Worktree").
    WorkingTree,
    /// A Worktree removed the way `git worktree remove` does: its own administrative entry
    /// under the Repo it was linked from, then its own working directory.
    Worktree,
    /// A Worktree whose parent Repo could not be opened, so its own working directory alone
    /// went, with no administrative entry to clean up.
    Directory,
}

impl Removal {
    /// The receipt's own first clause for this removal.
    fn said(self) -> &'static str {
        match self {
            Removal::WorkingTree => "working tree removed",
            Removal::Worktree => "worktree removed",
            Removal::Directory => "directory removed, its parent Repo was unreadable",
        }
    }
}

/// What `delete` did about the `[[repo]]` entry naming the path it removed. Three answers
/// rather than a flag: the write runs after the directory has gone, so it can fail with the
/// removal it describes already a fact.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum ConfigCleanup {
    EntryRemoved,
    NoEntryOfItsOwn,
    Failed(String),
}

/// What running the operation did to one row.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum Outcome {
    /// A `[[repo]]` entry now carries `exclude = true` for this path.
    Ignored,
    /// The `exclude` key is gone, and the entry with it if nothing else was left.
    Unignored,
    /// An `ignore` over an excluded row found no `[[repo]]` entry naming that entity's own
    /// path: its exclusion is inherited from an entry naming the git common dir it shares,
    /// which covers every entity sharing that dir
    /// ([config.md](../../../docs/spec/config.md)'s per-Repo entries). Removing that entry
    /// would show all of them again, which is not what this row asked for, so nothing is
    /// written and the row says so.
    ExcludedByAnInheritedEntry,
    /// A working tree `delete` confirmed gone, and what it then did about the `[[repo]]`
    /// entry naming the path. The removal is a fact by the time `config` is read, so a
    /// failing write is reported beside it rather than in place of it
    /// ([repo-management.md](../../../docs/spec/repo-management.md)'s "What `delete` leaves
    /// behind").
    Removed {
        removal: Removal,
        config: ConfigCleanup,
        /// What would not finish behind the removal: a linked Worktree the cascade could
        /// not take, or an administrative entry that would not clear. Empty on a removal
        /// that left nothing.
        problems: Vec<String>,
    },
    /// `sync` fast-forwarded the Repo's branch to its upstream.
    Synced,
    /// `sync` attempted the Repo and the auto-update's own five rules found it not eligible
    /// right now: eligibility can change between the gate and the run, so this is read only
    /// by attempting it, never a gate refusal
    /// ([repo-management.md](../../../docs/spec/repo-management.md)'s "What `sync` refuses,
    /// and why").
    NotEligibleToSync(SyncIneligibility),
    /// The gate already named this one and counted it; it is carried through so the report
    /// after the run names it too.
    Refused(Refusal),
    /// The operation was attempted and did not finish: a working tree that would not remove,
    /// or a config file that would not write.
    Failed(String),
    /// `before_sync` named a hook and one of its steps failed, so `sync` was never attempted
    /// ([repo-management.md](../../../docs/spec/repo-management.md)'s "Hooks around sync").
    BeforeSyncHookFailed(String),
    /// `sync` fast-forwarded the Repo, but `after_sync` named a hook and one of its steps
    /// failed. The fast-forward already happened and is never undone.
    SyncedAfterHookFailed(String),
}

/// What running a resolved `before_sync` or `after_sync` hook against one Repo found, once
/// its steps have finished: `None` when the Set active for the run names no hook at all, so
/// `sync_one` never has to distinguish "no hook" from "a hook that passed" itself.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum HookOutcome {
    Passed,
    Failed(String),
}

/// [`HookOutcome`] from a hook's own [`repon_core::ActionReceipt`]: the first step that
/// failed, in the step's own words, or [`HookOutcome::Passed`] when every step ran clean.
/// [`repon_core::StepOutcome::is_failure`] is the identical classification the fold and the
/// `action:` Filter term already share, so a hook and a configured `[[action]]` never
/// disagree about what counts as failing.
pub(crate) fn hook_outcome_from_receipt(receipt: &repon_core::ActionReceipt) -> HookOutcome {
    match receipt.steps.iter().find(|step| step.outcome.is_failure()) {
        Some(step) => HookOutcome::Failed(describe_step_failure(step)),
        None => HookOutcome::Passed,
    }
}

/// One failed [`repon_core::StepResult`] in a sentence: its own label, then what it exited
/// with or, for a step Repon performed itself, its own words.
fn describe_step_failure(step: &repon_core::StepResult) -> String {
    match &step.outcome {
        repon_core::StepOutcome::Failed(code) => format!("`{}` exited {code}", step.label),
        repon_core::StepOutcome::OwnWork(own_work) => {
            format!("`{}` {}", step.label, own_work.said())
        }
        repon_core::StepOutcome::Ok
        | repon_core::StepOutcome::NotRun
        | repon_core::StepOutcome::Cancelled => {
            format!("`{}` did not run to a failing exit", step.label)
        }
    }
}

/// Which of the fast-forward-only auto-update's own five rules found a Repo not eligible for
/// `sync` right now, reused unchanged from [`repon_core::AutoUpdateAttempt`] rather than a
/// second vocabulary for the identical four reasons.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum SyncIneligibility {
    NotClean,
    NoUpstream,
    NotBehind,
    NotFastForward,
}

impl SyncIneligibility {
    /// The reason the receipt and the log line give, in the auto-update's own terms.
    pub(crate) fn reason(self) -> &'static str {
        match self {
            SyncIneligibility::NotClean => "the working tree or index carries a change of its own",
            SyncIneligibility::NoUpstream => "no branch, no remote, or no upstream configured",
            SyncIneligibility::NotBehind => "already level with its upstream",
            SyncIneligibility::NotFastForward => {
                "the local branch has a commit its upstream does not"
            }
        }
    }
}

/// One outcome as the receipt records it: which grade of work Repon did, and its own words
/// for it (repo-management.md's "Receipts" table). Exhaustive over [`Outcome`], so a seventh
/// outcome has to say which grade it earns rather than inheriting one.
///
/// A refusal and an unchanged row are `Refused` rather than failures: nothing went wrong, so
/// neither may put a `!` in the gutter of a Repo that reads perfectly well.
pub(crate) fn own_work(outcome: &Outcome) -> OwnWork {
    match outcome {
        Outcome::Ignored => OwnWork::Did(Arc::from("ignored")),
        Outcome::Unignored => OwnWork::Did(Arc::from("no longer ignored")),
        Outcome::Removed {
            removal,
            config,
            problems,
        } => OwnWork::Did(Arc::from(removed_words(*removal, config, problems))),
        Outcome::ExcludedByAnInheritedEntry => OwnWork::Refused(Arc::from(
            "still ignored: the `[[repo]]` entry excluding it names another path",
        )),
        Outcome::Synced => OwnWork::Did(Arc::from("fast-forwarded to its upstream")),
        Outcome::NotEligibleToSync(reason) => OwnWork::Refused(Arc::from(format!(
            "not eligible to sync, {}",
            reason.reason()
        ))),
        Outcome::Refused(refusal) => {
            OwnWork::Refused(Arc::from(format!("refused, {}", refusal.reason())))
        }
        Outcome::Failed(error) => OwnWork::CouldNotAct(Arc::from(format!("failed, {error}"))),
        Outcome::BeforeSyncHookFailed(error) => OwnWork::CouldNotAct(Arc::from(format!(
            "before_sync hook failed, sync was not attempted: {error}"
        ))),
        Outcome::SyncedAfterHookFailed(error) => OwnWork::Did(Arc::from(format!(
            "fast-forwarded to its upstream; after_sync hook failed: {error}"
        ))),
    }
}

/// One removal's own sentence: what went, then what became of the `[[repo]]` entry naming
/// it, then whatever would not finish behind it. Anything that failed is named after the
/// removal rather than instead of it.
fn removed_words(removal: Removal, config: &ConfigCleanup, problems: &[String]) -> String {
    let said = match config {
        ConfigCleanup::EntryRemoved => format!("{}, `[[repo]]` entry removed", removal.said()),
        ConfigCleanup::NoEntryOfItsOwn => {
            format!("{}, no `[[repo]]` entry of its own", removal.said())
        }
        ConfigCleanup::Failed(error) => format!(
            "{}; its `[[repo]]` entry could not be removed: {error}",
            removal.said()
        ),
    };
    with_problems(said, problems)
}

/// `said` with each thing that would not finish named after it, the one place a receipt
/// appends them so a removal and a failure read the same way.
fn with_problems(mut said: String, problems: &[String]) -> String {
    for problem in problems {
        said.push_str("; ");
        said.push_str(problem);
    }
    said
}

/// One outcome as a sentence, for the log line each row gets after a run. Read out of
/// [`own_work`] rather than written a second time, so the log and the detail pane always say
/// the same thing about the same row.
pub(crate) fn describe(outcome: &Outcome) -> String {
    own_work(outcome).said().to_string()
}

/// One row: which Entity, its name, what happened to it, what its work confirmed gone, and
/// how long that took.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct Record {
    pub(crate) key: EntityKey,
    pub(crate) name: Arc<str>,
    pub(crate) outcome: Outcome,
    /// Every Entity whose working directory this row's own work confirmed gone: the row
    /// itself for a `delete` that removed it, plus each linked Worktree the cascade took
    /// with it. Separate from `outcome` because a confirmed removal is a fact about the
    /// filesystem and an outcome is a verdict on the selected row, so a row that failed
    /// still carries whatever its cascade had already taken, and a row that removed
    /// nothing carries none.
    pub(crate) removed: Vec<EntityKey>,
    /// What the act itself took. Real rather than nominal: `delete` walks a whole working
    /// tree, which is the one management operation that can visibly stall.
    pub(crate) elapsed: Duration,
}

/// What a whole run did, per row, for the caller to announce and log.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct Report {
    pub(crate) operation: Operation,
    pub(crate) records: Vec<Record>,
}

impl Report {
    /// Every row as [`repon_core::Core::record_own_work`] takes it: the Entity, the grade of
    /// work Repon did with its own words, and how long the act took. The receipt's words come
    /// from [`own_work`], the same place the log line reads.
    pub(crate) fn own_work_records(&self) -> Vec<(EntityKey, OwnWork, Duration)> {
        self.records
            .iter()
            .map(|record| {
                (
                    record.key.clone(),
                    own_work(&record.outcome),
                    record.elapsed,
                )
            })
            .collect()
    }

    /// Every Entity this run confirmed gone, for the caller to drop from the table itself
    /// ([repo-management.md](../../../docs/spec/repo-management.md)'s "What `delete` leaves
    /// behind"). Repon caused these absences, so they never become Vanished, which asks the
    /// user to acknowledge one it did not cause. Read off each row's own confirmed removals
    /// rather than off its outcome, so a Repo's row brings the linked Worktrees its cascade
    /// took with it and a row whose working tree is, or may still be, on disk brings
    /// nothing.
    pub(crate) fn removed_keys(&self) -> Vec<EntityKey> {
        self.records
            .iter()
            .flat_map(|record| record.removed.iter().cloned())
            .collect()
    }

    /// The one-line summary a Notice carries: the counts, never a silent success.
    pub(crate) fn summary(&self) -> String {
        let mut done = 0usize;
        let mut refused = 0usize;
        let mut unchanged = 0usize;
        let mut not_eligible = 0usize;
        let mut failed = 0usize;
        let mut after_hook_failed = 0usize;
        let mut cleanup_unfinished = 0usize;
        for record in &self.records {
            if matches!(record.outcome, Outcome::SyncedAfterHookFailed(_)) {
                after_hook_failed += 1;
            }
            match &record.outcome {
                Outcome::Ignored
                | Outcome::Unignored
                | Outcome::Synced
                | Outcome::SyncedAfterHookFailed(_) => done += 1,
                Outcome::Removed {
                    config, problems, ..
                } => {
                    done += 1;
                    if matches!(config, ConfigCleanup::Failed(_)) || !problems.is_empty() {
                        cleanup_unfinished += 1;
                    }
                }
                Outcome::ExcludedByAnInheritedEntry => unchanged += 1,
                Outcome::NotEligibleToSync(_) => not_eligible += 1,
                Outcome::Refused(_) => refused += 1,
                Outcome::Failed(_) | Outcome::BeforeSyncHookFailed(_) => failed += 1,
            }
        }
        let mut parts = vec![format!("{done} done")];
        if refused > 0 {
            parts.push(format!("{refused} refused"));
        }
        if unchanged > 0 {
            parts.push(format!("{unchanged} still ignored by another entry"));
        }
        if not_eligible > 0 {
            parts.push(format!("{not_eligible} not eligible to sync"));
        }
        if failed > 0 {
            parts.push(format!("{failed} failed"));
        }
        if after_hook_failed > 0 {
            parts.push(format!("{after_hook_failed} after_sync hook failed"));
        }
        if cleanup_unfinished > 0 {
            parts.push(format!(
                "{cleanup_unfinished} removed with cleanup unfinished"
            ));
        }
        format!("{}: {}", self.operation.name(), parts.join(", "))
    }
}

/// The Notice a management run raises for itself when `Action::Unwind` stopped it between
/// rows rather than letting it reach its own end: [`Report::summary`]'s own counts, prefixed
/// with how many of the whole Selection the run actually reached, so the truncation is
/// explicit rather than a summary that quietly undercounts what a full run would have shown.
/// `total` is the run's own starting size, from before the cancellation, never
/// `report.records.len()` again.
pub(crate) fn cancelled_summary(report: &Report, total: usize) -> String {
    format!(
        "{}, cancelled after {}/{total}",
        report.summary(),
        report.records.len()
    )
}

/// Runs `plan`'s operation against one `target` alone, timing the act and turning its
/// outcome into a `Record`. Exposed separately from [`run`] so a caller that wants to act
/// between rows, publishing each one's own position as it starts and checking for a
/// cancellation ahead of the next, can drive the Selection's own loop itself rather than
/// losing control to `run` for the whole thing at once; `run` itself is this called once per
/// target, in order, with nothing between calls.
///
/// `worktree_admin_dir`, `linked_worktree_paths`, `ignored_directories_for_deletion` and
/// `attempt_sync` are [`repon_core::ManagementHandle::worktree_admin_dir`],
/// [`repon_core::ManagementHandle::linked_worktree_paths`],
/// [`repon_core::ManagementHandle::ignored_directories_for_deletion`] and
/// [`repon_core::ManagementHandle::attempt_auto_update`] at the one call site; taken as
/// parameters, the same way [`Plan::with_risk`] takes `read`, so this module never needs a
/// `Core` to be tested. `run_before_sync_hook` and `run_after_sync_hook` are consulted for
/// `Operation::Sync` alone, `None` meaning the Set active for this run names no hook at all
/// ([repo-management.md](../../../docs/spec/repo-management.md)'s "Hooks around sync").
#[allow(clippy::too_many_arguments)]
pub(crate) fn run_one_record(
    plan: &Plan,
    target: &Target,
    config_file: &Path,
    worktree_admin_dir: impl Fn(&EntityKey) -> Option<PathBuf>,
    linked_worktree_paths: impl Fn(&EntityKey) -> Vec<PathBuf>,
    ignored_directories_for_deletion: impl Fn(&Path) -> Vec<PathBuf>,
    attempt_sync: impl Fn(&EntityKey) -> AutoUpdateAttempt,
    run_before_sync_hook: impl Fn(&EntityKey) -> Option<HookOutcome>,
    run_after_sync_hook: impl Fn(&EntityKey) -> Option<HookOutcome>,
) -> Record {
    let started = Instant::now();
    let (outcome, removed) = match target.eligibility {
        Eligibility::Refused(refusal) => (Outcome::Refused(refusal), Vec::new()),
        Eligibility::Eligible => run_one(
            plan.operation,
            target,
            config_file,
            &worktree_admin_dir,
            &linked_worktree_paths,
            &ignored_directories_for_deletion,
            &attempt_sync,
            &run_before_sync_hook,
            &run_after_sync_hook,
        )
        .unwrap_or_else(|err| (Outcome::Failed(format!("{err:#}")), Vec::new())),
    };
    Record {
        key: target.key.clone(),
        name: Arc::clone(&target.name),
        outcome,
        removed,
        elapsed: started.elapsed(),
    }
}

/// Runs `plan` against `config_file`, in the Selection's own order, and reports what happened
/// to every row including the ones the gate already refused.
///
/// `config_file` is passed in rather than resolved here, so a test drives this against a
/// temp directory of its own making and never against the process-wide path
/// [`crate::config::config_file`] fixes. Every row is [`run_one_record`], called once per
/// target with nothing of this function's own between calls; a caller that needs to act
/// between rows drives `run_one_record` itself instead.
#[allow(clippy::too_many_arguments)]
#[allow(dead_code)] // exercised by this module's own tests; production drives run_one_record
// itself, to repaint between rows
pub(crate) fn run(
    plan: &Plan,
    config_file: &Path,
    worktree_admin_dir: impl Fn(&EntityKey) -> Option<PathBuf>,
    linked_worktree_paths: impl Fn(&EntityKey) -> Vec<PathBuf>,
    ignored_directories_for_deletion: impl Fn(&Path) -> Vec<PathBuf>,
    attempt_sync: impl Fn(&EntityKey) -> AutoUpdateAttempt,
    run_before_sync_hook: impl Fn(&EntityKey) -> Option<HookOutcome>,
    run_after_sync_hook: impl Fn(&EntityKey) -> Option<HookOutcome>,
) -> Report {
    let records = plan
        .targets
        .iter()
        .map(|target| {
            run_one_record(
                plan,
                target,
                config_file,
                &worktree_admin_dir,
                &linked_worktree_paths,
                &ignored_directories_for_deletion,
                &attempt_sync,
                &run_before_sync_hook,
                &run_after_sync_hook,
            )
        })
        .collect();
    Report {
        operation: plan.operation,
        records,
    }
}

#[allow(clippy::too_many_arguments)]
fn run_one(
    operation: Operation,
    target: &Target,
    config_file: &Path,
    worktree_admin_dir: &impl Fn(&EntityKey) -> Option<PathBuf>,
    linked_worktree_paths: &impl Fn(&EntityKey) -> Vec<PathBuf>,
    ignored_directories_for_deletion: &impl Fn(&Path) -> Vec<PathBuf>,
    attempt_sync: &impl Fn(&EntityKey) -> AutoUpdateAttempt,
    run_before_sync_hook: &impl Fn(&EntityKey) -> Option<HookOutcome>,
    run_after_sync_hook: &impl Fn(&EntityKey) -> Option<HookOutcome>,
) -> Result<(Outcome, Vec<EntityKey>)> {
    match operation {
        Operation::Ignore if target.excluded => {
            if repo_entry::write(config_file, target.key.path(), Edit::Unexclude)?.changed {
                Ok((Outcome::Unignored, Vec::new()))
            } else {
                Ok((Outcome::ExcludedByAnInheritedEntry, Vec::new()))
            }
        }
        Operation::Ignore => {
            repo_entry::write(config_file, target.key.path(), Edit::Exclude)?;
            Ok((Outcome::Ignored, Vec::new()))
        }
        Operation::Delete => delete_one(
            target,
            config_file,
            worktree_admin_dir,
            linked_worktree_paths,
            ignored_directories_for_deletion,
        ),
        Operation::Sync => Ok((
            sync_one(
                target,
                attempt_sync,
                run_before_sync_hook,
                run_after_sync_hook,
            ),
            Vec::new(),
        )),
    }
}

/// `sync` on one row: `attempt_sync` is called only for the Repos [`Operation::eligibility`]
/// already found eligible by Kind, and its own answer is reported rather than fixed, per
/// [repo-management.md](../../../docs/spec/repo-management.md)'s "What `sync` refuses, and
/// why": an ineligible-right-now Repo is not a failure, so it never reaches [`Outcome::Failed`].
///
/// `run_before_sync_hook` runs, and is checked, before `attempt_sync` is ever called: a
/// failing pre-hook means the fast-forward is never attempted for this row at all.
/// `run_after_sync_hook` runs only once `attempt_sync` reports [`AutoUpdateAttempt::Updated`],
/// and its own failure never undoes that fast-forward, which already happened
/// ([repo-management.md](../../../docs/spec/repo-management.md)'s "Hooks around sync").
fn sync_one(
    target: &Target,
    attempt_sync: &impl Fn(&EntityKey) -> AutoUpdateAttempt,
    run_before_sync_hook: &impl Fn(&EntityKey) -> Option<HookOutcome>,
    run_after_sync_hook: &impl Fn(&EntityKey) -> Option<HookOutcome>,
) -> Outcome {
    if let Some(HookOutcome::Failed(error)) = run_before_sync_hook(&target.key) {
        return Outcome::BeforeSyncHookFailed(error);
    }
    let outcome = match attempt_sync(&target.key) {
        AutoUpdateAttempt::Updated => Outcome::Synced,
        AutoUpdateAttempt::NotClean => Outcome::NotEligibleToSync(SyncIneligibility::NotClean),
        AutoUpdateAttempt::NoUpstream => Outcome::NotEligibleToSync(SyncIneligibility::NoUpstream),
        AutoUpdateAttempt::NotBehind => Outcome::NotEligibleToSync(SyncIneligibility::NotBehind),
        AutoUpdateAttempt::NotFastForward => {
            Outcome::NotEligibleToSync(SyncIneligibility::NotFastForward)
        }
        AutoUpdateAttempt::Failed(error) => Outcome::Failed(error),
    };
    if matches!(outcome, Outcome::Synced)
        && let Some(HookOutcome::Failed(error)) = run_after_sync_hook(&target.key)
    {
        return Outcome::SyncedAfterHookFailed(error);
    }
    outcome
}

/// The worker count [`delete_ignored_directories`] bounds its own pool to: a fixed constant
/// rather than `available_parallelism`, since the right number for metadata-heavy deletion on
/// a given filesystem is a measured fact, not one the OS's own core count states
/// ([repo-management.md](../../../docs/spec/repo-management.md)'s "Deleting a working tree").
const IGNORED_DIRECTORY_DELETE_WORKERS: usize = 4;

/// `delete`'s phase 2: deletes each of `directories` independently on a bounded rayon pool
/// built and torn down for this call alone, the same shape
/// `repon_core`'s own periodic-fetch `run_bounded` uses for the identical reason: this must
/// never take a worker away from rayon's global pool, where every probe already lives.
///
/// Returns the directories that would not remove, rather than stopping at the first one: a
/// stuck permission bit on one large ignored subtree must not hold back the rest, and phase 3
/// ([`remove_working_tree`]) still runs against whatever this call leaves behind, so a failure
/// here is never fatal to `delete` as a whole.
fn delete_ignored_directories(directories: Vec<PathBuf>) -> Vec<PathBuf> {
    use rayon::iter::{IntoParallelIterator, ParallelIterator};

    if directories.is_empty() {
        return Vec::new();
    }
    let pool = rayon::ThreadPoolBuilder::new()
        .num_threads(IGNORED_DIRECTORY_DELETE_WORKERS)
        .build()
        .expect("build the ignored-directory delete pool");
    pool.install(|| {
        directories
            .into_par_iter()
            .filter(|dir| fs::remove_dir_all(dir).is_err())
            .collect()
    })
}

/// `delete` on one row: a Repo takes its linked Worktrees' own directories with it, a
/// Worktree is removed the way `git worktree remove` does when its parent Repo can still be
/// opened and falls back to a bare directory removal when it cannot, and a Submodule never
/// reaches here at all, since [`Operation::eligibility`] always refuses it first.
///
/// Each working tree removed here goes through two phases first, per
/// [repo-management.md](../../../docs/spec/repo-management.md)'s "Deleting a working
/// tree": `ignored_directories_for_deletion` enumerates that tree's own ignored
/// directories (phase 1), [`delete_ignored_directories`] drains them independently (phase 2),
/// and only then does [`remove_working_tree`] run (phase 3, unchanged from before this
/// existed). Phase 2 deletes a strict subset of what phase 3 deletes anyway, so a crash, or a
/// phase-2 failure this function ignores, between the two leaves nothing this row's own
/// `delete` cannot still finish.
fn delete_one(
    target: &Target,
    config_file: &Path,
    worktree_admin_dir: &impl Fn(&EntityKey) -> Option<PathBuf>,
    linked_worktree_paths: &impl Fn(&EntityKey) -> Vec<PathBuf>,
    ignored_directories_for_deletion: &impl Fn(&Path) -> Vec<PathBuf>,
) -> Result<(Outcome, Vec<EntityKey>)> {
    match target.kind {
        Kind::Repo => {
            let mut removed = Vec::new();
            let mut problems = Vec::new();
            for worktree in linked_worktree_paths(&target.key) {
                // Read before the directory goes, since resolving needs it on disk.
                let key = worktree_key(&worktree);
                delete_ignored_directories(ignored_directories_for_deletion(&worktree));
                // Best effort: a sibling Worktree that will not remove never stops the
                // Repo's own removal below. Only the ones that did go are staged, and the
                // ones that did not are named, so the report claims no directory that is
                // still on disk.
                match remove_working_tree(&worktree) {
                    Ok(()) => removed.push(key),
                    Err(err) => problems.push(format!(
                        "its linked Worktree at {} would not remove: {err:#}",
                        worktree.display()
                    )),
                }
            }
            delete_ignored_directories(ignored_directories_for_deletion(target.key.path()));
            if let Err(err) = remove_working_tree(target.key.path()) {
                // What the cascade already took is a fact whatever becomes of the Repo's
                // own tree, so it is reported rather than thrown away with the error: a
                // directory that is gone must not be left with a row pointing at it.
                return Ok((
                    Outcome::Failed(with_problems(format!("{err:#}"), &problems)),
                    removed,
                ));
            }
            removed.push(target.key.clone());
            let config = clean_up_config(config_file, target.key.path());
            Ok((
                Outcome::Removed {
                    removal: Removal::WorkingTree,
                    config,
                    problems,
                },
                removed,
            ))
        }
        Kind::Worktree => {
            // Read before either removal runs: once the working tree is gone, its own
            // `.git` file is gone with it and the admin dir can no longer be found.
            let admin_dir = worktree_admin_dir(&target.key);
            delete_ignored_directories(ignored_directories_for_deletion(target.key.path()));
            // The working tree first, the admin dir second: the same order
            // `git worktree remove` itself uses, so a failure part-way through (a
            // permissions error inside `remove_dir_all`, say) leaves the parent still
            // knowing about a Worktree whose own directory is gone, never the reverse:
            // a directory on disk the parent has already forgotten and whose `.git`
            // file now dangles, which `git worktree repair` cannot fix.
            remove_working_tree(target.key.path())?;
            let mut problems = Vec::new();
            // An entry the parent Repo has already pruned is nothing to clear, so only an
            // entry that is there and refuses is named.
            if let Some(admin_dir) = &admin_dir
                && let Err(err) = fs::remove_dir_all(admin_dir)
                && err.kind() != std::io::ErrorKind::NotFound
            {
                problems.push(format!(
                    "its administrative entry under its parent Repo would not clear: {err}"
                ));
            }
            let config = clean_up_config(config_file, target.key.path());
            let removal = if admin_dir.is_some() {
                Removal::Worktree
            } else {
                Removal::Directory
            };
            Ok((
                Outcome::Removed {
                    removal,
                    config,
                    problems,
                },
                vec![target.key.clone()],
            ))
        }
        Kind::Submodule => {
            unreachable!("a Submodule is always refused before `delete` reaches a row")
        }
    }
}

/// The key naming the linked Worktree at `path`, resolved the way discovery resolves every
/// key it makes: git's own register need not spell the directory that way, and a key that
/// matches no row is a row nothing drops. Falls back to the register's own spelling when the
/// path will not resolve, which is the best guess left.
fn worktree_key(path: &Path) -> EntityKey {
    let resolved = path.canonicalize();
    EntityKey::new(Arc::from(resolved.as_deref().unwrap_or(path)))
}

/// `delete`'s config half, run once the working tree is gone: the `[[repo]]` entry naming
/// the removed path, and that path from every `[[set]]` array naming it
/// ([repo-management.md](../../../docs/spec/repo-management.md)'s "Writing config"). A write
/// that fails is reported rather than propagated, since the directory it describes has
/// already gone and the row it belongs to is removed either way.
fn clean_up_config(config_file: &Path, path: &Path) -> ConfigCleanup {
    match repo_entry::write(config_file, path, Edit::Remove) {
        Ok(written) if written.removed_repo_entry => ConfigCleanup::EntryRemoved,
        Ok(_) => ConfigCleanup::NoEntryOfItsOwn,
        Err(err) => ConfigCleanup::Failed(format!("{err:#}")),
    }
}

/// Removes one Repo's or Worktree's working tree, the whole directory `path` names.
///
/// The path comes from the key discovery resolved, or from git's own worktree register for
/// a cascading Repo delete, never from config, an environment variable or the working
/// directory. The two guards below can only refuse: a relative path is one neither source
/// ever produces (an [`EntityKey`] is a resolved absolute working directory, and the path
/// git's own register hands back is absolute however the register itself spelled it), and a
/// directory with no `.git` in it is not the Repo or
/// Worktree this call named, so either means something other than the intended one is about
/// to be removed permanently.
fn remove_working_tree(path: &Path) -> Result<()> {
    if !path.is_absolute() {
        return Err(eyre!(
            "refusing to delete a relative path: {}",
            path.display()
        ));
    }
    if !path.join(".git").exists() {
        return Err(eyre!(
            "refusing to delete {}: no `.git` there, so it is not the Repo this row named",
            path.display()
        ));
    }
    fs::remove_dir_all(path).map_err(|err| eyre!("could not remove {}: {err}", path.display()))?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use repon_core::EntityKey;
    use std::path::PathBuf;

    fn spec_source() -> String {
        std::fs::read_to_string(
            std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
                .join("../../docs/spec/repo-management.md"),
        )
        .expect("read docs/spec/repo-management.md")
    }

    fn entity(path: &Path, name: &str, kind: Kind) -> EntityState {
        let path: Arc<Path> = Arc::from(path);
        EntityState::new(
            EntityKey::new(Arc::clone(&path)),
            Arc::from(name),
            path,
            kind,
        )
    }

    fn excluded(mut entity: EntityState) -> EntityState {
        entity.excluded = true;
        entity
    }

    /// Every entity as a checked Selection, which is what a plan built over all of them
    /// means in these tests.
    fn checked(entities: &[EntityState]) -> Targets {
        Targets {
            keys: entities.iter().map(|entity| entity.key.clone()).collect(),
            scope: RunScope::CheckedRows,
        }
    }

    fn plan(operation: Operation, entities: &[EntityState]) -> Plan {
        Plan::new(operation, entities, checked(entities))
    }

    /// A Worktree sharing `parent`'s own path as its `common_dir`, the fixture's stand-in
    /// for two Entities attached to the same Repo: real discovery ties them by the git
    /// common dir they share, and every dedup this module does (`delete` merging a Worktree
    /// into its selected parent's own removal) reads that same field.
    fn worktree_of(parent: &EntityState, path: &Path, name: &str) -> EntityState {
        EntityState::new(
            EntityKey::new(Arc::from(path)),
            Arc::from(name),
            Arc::clone(&parent.common_dir),
            Kind::Worktree,
        )
    }

    /// [`run`] with no Worktree of its own to remove and no `sync` attempt to make: every
    /// test that is not itself about the Worktree-removal cascade or `sync` wires trivial
    /// closures here rather than repeating them.
    fn run_plain(plan: &Plan, config_file: &Path) -> Report {
        run(
            plan,
            config_file,
            |_| None,
            |_| Vec::new(),
            |_| Vec::new(),
            |_| panic!("run_plain does not exercise sync"),
            |_| panic!("run_plain declares no before_sync hook"),
            |_| panic!("run_plain declares no after_sync hook"),
        )
    }

    /// [`Outcome::Removed`] with nothing left behind, the shape every fixture that is not
    /// itself about an unfinished cleanup produces.
    fn removed(removal: Removal, config: ConfigCleanup) -> Outcome {
        Outcome::Removed {
            removal,
            config,
            problems: Vec::new(),
        }
    }

    /// The three names, and their order, come from repo-management.md's own operations table
    /// read at test time, never restated here: the reserved-name check in
    /// [`crate::config::document`] and the palette's own built-in list are both this array,
    /// so a name that drifted from the specification would take both with it silently.
    #[test]
    fn the_built_in_names_are_repo_management_mds_own_operations_table() {
        let spec = spec_source();
        let table = spec
            .split("## The operations")
            .nth(1)
            .expect("the operations section is still there");
        let declared: Vec<String> = table
            .lines()
            .take_while(|line| line.starts_with('|') || line.trim().is_empty())
            .filter_map(|line| line.split('`').nth(1).map(str::to_string))
            .collect();

        assert_eq!(
            declared,
            OPERATIONS
                .iter()
                .map(|operation| operation.name().to_string())
                .collect::<Vec<_>>(),
            "the compiled built-ins must be exactly the specification's own operations, in \
             its own order"
        );
    }

    /// The per-row Notice `App::run_management` paints before each row's own work starts:
    /// the operation, the row's own name, and its position among every row this run visits,
    /// refused rows counted too since the run still visits them in order.
    #[test]
    fn row_notice_names_the_operation_the_row_and_its_position() {
        assert_eq!(
            row_notice(Operation::Delete, "manage-pr-1358", 3, 12),
            "delete: manage-pr-1358 (3/12)"
        );
    }

    // =====================================================================================
    // Criterion 5: `delete` is refused on a Submodule alone; a Worktree is eligible.
    // =====================================================================================

    #[test]
    fn delete_is_refused_on_a_submodule_and_it_is_named_and_counted() {
        let entities = vec![
            entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo),
            entity(Path::new("/tmp/x/sub"), "sub", Kind::Submodule),
        ];

        let plan = plan(Operation::Delete, &entities);

        assert_eq!(plan.eligible_count(), 1, "only the Repo is eligible");
        assert_eq!(plan.refused_count(), 1, "and the refusal is counted");

        let lines = plan.confirm_lines();
        assert!(
            lines[0].contains('1') && lines[0].contains("1 refused"),
            "the headline must carry both counts, got {:?}",
            lines[0]
        );
        let line = lines
            .iter()
            .find(|line| line.starts_with("sub"))
            .unwrap_or_else(|| panic!("no line names sub in {lines:?}"));
        assert!(
            line.contains("refused") && line.contains(Refusal::SubmoduleCannotBeDeleted.reason()),
            "a refusal must name itself and say why, got {line:?}"
        );
    }

    /// A Worktree with no selected parent is eligible for `delete` on its own, the scope
    /// rule this ticket overrules.
    #[test]
    fn a_worktree_is_eligible_for_delete_when_its_parent_is_not_also_selected() {
        let repo = entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo);
        let tree = worktree_of(&repo, Path::new("/tmp/x/tree"), "tree");

        let entities = [tree];
        let plan = Plan::new(Operation::Delete, &entities, checked(&entities));

        assert_eq!(
            plan.eligible_count(),
            1,
            "the Worktree is eligible on its own"
        );
        assert_eq!(plan.targets.len(), 1);
        assert_eq!(plan.targets[0].eligibility, Eligibility::Eligible);
    }

    /// A refusal is not merely a line, it is a row nothing happens to. Every directory here
    /// is created by this test in a temp directory of its own making; no path comes from
    /// config, an environment variable or the working directory.
    #[test]
    fn running_delete_removes_the_repo_alone_and_leaves_the_refused_submodule_on_disk() {
        let dir = tempfile::tempdir().expect("temp dir");
        let config_file = dir.path().join("config.toml");
        let made = |name: &str| -> PathBuf {
            let path = dir.path().join(name);
            std::fs::create_dir_all(path.join(".git")).expect("create a fixture directory");
            path
        };
        let repo = made("repo");
        let sub = made("sub");
        let entities = vec![
            entity(&repo, "repo", Kind::Repo),
            entity(&sub, "sub", Kind::Submodule),
        ];

        let report = run_plain(&plan(Operation::Delete, &entities), &config_file);

        assert!(!repo.exists(), "the Repo's working tree is gone");
        assert!(sub.exists(), "a Submodule is never removed");
        assert_eq!(
            report
                .records
                .iter()
                .map(|record| (record.name.to_string(), record.outcome.clone()))
                .collect::<Vec<_>>(),
            vec![
                (
                    "repo".to_string(),
                    removed(Removal::WorkingTree, ConfigCleanup::NoEntryOfItsOwn)
                ),
                (
                    "sub".to_string(),
                    Outcome::Refused(Refusal::SubmoduleCannotBeDeleted)
                ),
            ],
            "every row is reported, the refusal included"
        );
        assert!(
            report.summary().contains("1 refused"),
            "the summary must count the refusal rather than announce a clean run, got {:?}",
            report.summary()
        );
    }

    /// A row a Set names but no `[[repo]]` entry does: the path leaves the Set's own array,
    /// and the receipt still says there was no entry of its own, since dropping a path from
    /// a Set is not an entry going with the working tree.
    #[test]
    fn deleting_a_repo_named_only_by_a_set_says_there_was_no_entry_of_its_own() {
        let dir = tempfile::tempdir().expect("temp dir");
        let config_file = dir.path().join("config.toml");
        let repo = dir.path().join("repo");
        std::fs::create_dir_all(repo.join(".git")).expect("create a fixture directory");
        std::fs::write(
            &config_file,
            format!(
                "[[set]]\nname = \"one\"\nroots = [\"{root}\"]\ninclude = [\"{repo}\", \
                 \"**/kept/**\"]\n",
                root = dir.path().display(),
                repo = repo.display(),
            ),
        )
        .expect("write config.toml");
        let entities = vec![entity(&repo, "repo", Kind::Repo)];

        let report = run_plain(&plan(Operation::Delete, &entities), &config_file);

        assert_eq!(
            report.records[0].outcome,
            removed(Removal::WorkingTree, ConfigCleanup::NoEntryOfItsOwn),
            "the Set naming it is not a `[[repo]]` entry of its own"
        );
        let written = std::fs::read_to_string(&config_file).expect("read config.toml back");
        assert!(
            !written.contains(&repo.display().to_string()) && written.contains("**/kept/**"),
            "the Set stops naming the deleted path and keeps its glob: {written:?}"
        );
    }

    // =====================================================================================
    // `delete` on a Worktree: removed the way `git worktree remove` does when its parent
    // can still be found, falling back to a bare directory removal when it cannot. The
    // admin-dir and linked-worktree-paths closures stand in for `repon_core::Core`'s own
    // reads, taken as parameters so this module never needs a real git repository to test
    // this half either.
    // =====================================================================================

    #[test]
    fn deleting_a_worktree_removes_its_admin_dir_and_its_own_directory() {
        let dir = tempfile::tempdir().expect("temp dir");
        let config_file = dir.path().join("config.toml");
        let tree = dir.path().join("tree");
        std::fs::create_dir_all(tree.join(".git")).expect("create the worktree fixture");
        let admin_dir = dir.path().join("admin");
        std::fs::create_dir_all(&admin_dir).expect("create the admin dir fixture");
        let entities = vec![entity(&tree, "tree", Kind::Worktree)];

        let report = run(
            &plan(Operation::Delete, &entities),
            &config_file,
            |_| Some(admin_dir.clone()),
            |_| Vec::new(),
            |_| Vec::new(),
            |_| panic!("this test does not exercise sync"),
            |_| panic!("this test declares no before_sync hook"),
            |_| panic!("this test declares no after_sync hook"),
        );

        assert!(!tree.exists(), "the Worktree's own directory is gone");
        assert!(!admin_dir.exists(), "its administrative entry is gone too");
        assert_eq!(
            report.records[0].outcome,
            removed(Removal::Worktree, ConfigCleanup::NoEntryOfItsOwn)
        );
    }

    /// A Worktree whose own directory went but whose administrative entry under the parent
    /// would not clear: still a removal, and one the result tells apart from the clean
    /// `git worktree remove` it was not
    /// ([repo-management.md](../../../docs/spec/repo-management.md)'s "What `delete` does to
    /// a Worktree").
    #[test]
    fn a_worktree_whose_admin_entry_would_not_clear_is_told_apart_from_a_clean_removal() {
        let dir = tempfile::tempdir().expect("temp dir");
        let config_file = dir.path().join("config.toml");
        let tree = dir.path().join("tree");
        std::fs::create_dir_all(tree.join(".git")).expect("create the worktree fixture");
        // A file where the entry should be, so `remove_dir_all` refuses it every time
        // rather than racing a permission bit.
        let admin_dir = dir.path().join("admin-that-is-a-file");
        std::fs::write(&admin_dir, "not a directory").expect("create the admin fixture");
        let entities = vec![entity(&tree, "tree", Kind::Worktree)];

        let report = run(
            &plan(Operation::Delete, &entities),
            &config_file,
            |_| Some(admin_dir.clone()),
            |_| Vec::new(),
            |_| Vec::new(),
            |_| panic!("this test does not exercise sync"),
            |_| panic!("this test declares no before_sync hook"),
            |_| panic!("this test declares no after_sync hook"),
        );

        assert!(!tree.exists(), "the Worktree's own directory is still gone");
        assert!(
            matches!(
                &report.records[0].outcome,
                Outcome::Removed { removal, problems, .. }
                    if *removal == Removal::Worktree && problems.len() == 1
            ),
            "a removal whose administrative entry would not clear carries what it left, got \
             {:?}",
            report.records[0].outcome
        );
        let said = describe(&report.records[0].outcome);
        assert!(
            said.contains("its administrative entry under its parent Repo would not clear"),
            "and the receipt says what was left behind, got {said:?}"
        );
        assert_eq!(
            report.removed_keys(),
            vec![entities[0].key.clone()],
            "the directory went, so the row still leaves the table"
        );
    }

    /// A working tree that went and a `config.toml` write that did not are two facts, and
    /// the receipt carries both: the removal first, then the write that failed behind it,
    /// so the row never reads as the clean removal it was not
    /// ([repo-management.md](../../../docs/spec/repo-management.md)'s "What `delete` leaves
    /// behind").
    #[test]
    fn a_removal_whose_config_write_failed_names_the_write_after_the_removal() {
        let dir = tempfile::tempdir().expect("temp dir");
        // A directory where the file goes, so every read of the config path fails on its
        // first byte rather than racing a permission bit.
        let config_file = dir.path().join("config.toml");
        std::fs::create_dir(&config_file).expect("put a directory where the config file goes");
        let repo = dir.path().join("repo");
        std::fs::create_dir_all(repo.join(".git")).expect("create the repo fixture");
        let entities = vec![entity(&repo, "repo", Kind::Repo)];

        let report = run_plain(&plan(Operation::Delete, &entities), &config_file);

        assert!(!repo.exists(), "the working tree is genuinely gone");
        let said = describe(&report.records[0].outcome);
        assert!(
            said.starts_with("working tree removed; its `[[repo]]` entry could not be removed: "),
            "the receipt names the removal, then the write that did not finish, got {said:?}"
        );
        assert_eq!(
            report.removed_keys(),
            vec![entities[0].key.clone()],
            "the row leaves on the removal, never on the write"
        );
        assert_eq!(
            report.summary(),
            "delete: 1 done, 1 removed with cleanup unfinished",
            "and the completion carries both halves"
        );
    }

    /// An administrative entry that is already gone is nothing to clear rather than a
    /// cleanup that failed: the removal reads clean, and the completion does not count a
    /// row against work no one has left to do
    /// ([repo-management.md](../../../docs/spec/repo-management.md)'s "What `delete` does to
    /// a Worktree").
    #[test]
    fn a_worktree_whose_admin_entry_was_already_pruned_reads_as_a_clean_removal() {
        let dir = tempfile::tempdir().expect("temp dir");
        let config_file = dir.path().join("config.toml");
        let tree = dir.path().join("tree");
        std::fs::create_dir_all(tree.join(".git")).expect("create the worktree fixture");
        // What a dangling `.git` file names: an entry the parent Repo has already pruned.
        let admin_dir = dir.path().join("admin-that-was-never-there");
        let entities = vec![entity(&tree, "tree", Kind::Worktree)];

        let report = run(
            &plan(Operation::Delete, &entities),
            &config_file,
            |_| Some(admin_dir.clone()),
            |_| Vec::new(),
            |_| Vec::new(),
            |_| panic!("this test does not exercise sync"),
            |_| panic!("this test declares no before_sync hook"),
            |_| panic!("this test declares no after_sync hook"),
        );

        assert!(!tree.exists(), "the Worktree's own directory is gone");
        assert_eq!(
            report.records[0].outcome,
            removed(Removal::Worktree, ConfigCleanup::NoEntryOfItsOwn),
            "nothing was left behind, so nothing is named"
        );
        assert_eq!(
            report.summary(),
            "delete: 1 done",
            "and the completion counts one plain removal"
        );
    }

    /// The order matters, not just the end state: `git worktree remove` itself removes the
    /// working tree before the administrative entry, so a failure part-way leaves the
    /// parent still knowing about a Worktree whose own directory is gone, never the
    /// reverse. The two orders produce the same end state on the happy path and differ
    /// only when the working tree's own removal fails, which is what this pins.
    #[test]
    fn deleting_a_worktree_removes_the_working_tree_before_the_admin_dir() {
        let dir = tempfile::tempdir().expect("temp dir");
        let config_file = dir.path().join("config.toml");
        // No `.git` marker: `remove_working_tree`'s own guard refuses this path outright,
        // so the working tree's own removal fails before anything is deleted.
        let tree = dir.path().join("tree");
        std::fs::create_dir_all(&tree).expect("create a worktree fixture with no .git marker");
        let admin_dir = dir.path().join("admin");
        std::fs::create_dir_all(&admin_dir).expect("create the admin dir fixture");
        let entities = vec![entity(&tree, "tree", Kind::Worktree)];

        let report = run(
            &plan(Operation::Delete, &entities),
            &config_file,
            |_| Some(admin_dir.clone()),
            |_| Vec::new(),
            |_| Vec::new(),
            |_| panic!("this test does not exercise sync"),
            |_| panic!("this test declares no before_sync hook"),
            |_| panic!("this test declares no after_sync hook"),
        );

        assert!(
            matches!(report.records[0].outcome, Outcome::Failed(_)),
            "the working tree's own removal must fail first, got {:?}",
            report.records[0].outcome
        );
        assert!(
            admin_dir.exists(),
            "the admin dir must still be there: removing the working tree comes first, and \
             it never got the chance to succeed"
        );
    }

    /// The fallback: a Worktree whose parent cannot be opened is still removed, but only its
    /// own directory, and the report says so rather than claiming a clean `git worktree
    /// remove`.
    #[test]
    fn deleting_a_worktree_whose_parent_is_unreachable_falls_back_to_a_directory_removal() {
        let dir = tempfile::tempdir().expect("temp dir");
        let config_file = dir.path().join("config.toml");
        let tree = dir.path().join("tree");
        std::fs::create_dir_all(tree.join(".git")).expect("create the worktree fixture");
        let entities = vec![entity(&tree, "tree", Kind::Worktree)];

        let report = run(
            &plan(Operation::Delete, &entities),
            &config_file,
            |_| None,
            |_| Vec::new(),
            |_| Vec::new(),
            |_| panic!("this test does not exercise sync"),
            |_| panic!("this test declares no before_sync hook"),
            |_| panic!("this test declares no after_sync hook"),
        );

        assert!(!tree.exists(), "the Worktree's own directory is still gone");
        assert_eq!(
            report.records[0].outcome,
            removed(Removal::Directory, ConfigCleanup::NoEntryOfItsOwn)
        );
    }

    /// Deleting a Repo takes its linked Worktrees with it: each one's own directory sits
    /// outside the Repo's own and is removed too, read from the injected
    /// `linked_worktree_paths` the same way `repon_core::Core::linked_worktree_paths` would
    /// answer for the real thing.
    #[test]
    fn deleting_a_repo_removes_every_linked_worktrees_own_directory_too() {
        let dir = tempfile::tempdir().expect("temp dir");
        let config_file = dir.path().join("config.toml");
        let repo = dir.path().join("repo");
        std::fs::create_dir_all(repo.join(".git")).expect("create the repo fixture");
        let sibling_one = dir.path().join("sibling-one");
        let sibling_two = dir.path().join("sibling-two");
        std::fs::create_dir_all(sibling_one.join(".git")).expect("create sibling one");
        std::fs::create_dir_all(sibling_two.join(".git")).expect("create sibling two");
        let entities = vec![entity(&repo, "repo", Kind::Repo)];
        let siblings = [sibling_one.clone(), sibling_two.clone()];

        let report = run(
            &plan(Operation::Delete, &entities),
            &config_file,
            |_| None,
            |_| siblings.to_vec(),
            |_| Vec::new(),
            |_| panic!("this test does not exercise sync"),
            |_| panic!("this test declares no before_sync hook"),
            |_| panic!("this test declares no after_sync hook"),
        );

        assert!(!repo.exists(), "the Repo's own working tree is gone");
        assert!(!sibling_one.exists(), "the first linked Worktree is gone");
        assert!(!sibling_two.exists(), "the second linked Worktree is gone");
        assert_eq!(
            report.records[0].outcome,
            removed(Removal::WorkingTree, ConfigCleanup::NoEntryOfItsOwn)
        );
    }

    /// A Repo whose own working tree would not remove after its cascade already took a
    /// linked Worktree: what went is still reported, so the rows over those directories
    /// leave the table rather than being left to become Vanished
    /// ([repo-management.md](../../../docs/spec/repo-management.md)'s "What `delete` leaves
    /// behind").
    #[test]
    fn a_repo_whose_own_removal_fails_still_reports_what_its_cascade_took() {
        let dir = tempfile::tempdir().expect("temp dir");
        let root = dir
            .path()
            .canonicalize()
            .expect("canonicalize the temp dir");
        let config_file = root.join("config.toml");
        // No `.git` in it, so `remove_working_tree`'s own guard refuses the Repo once the
        // cascade has already run.
        let repo = root.join("repo");
        std::fs::create_dir_all(&repo).expect("create the repo fixture");
        let sibling = root.join("sibling");
        std::fs::create_dir_all(sibling.join(".git")).expect("create the linked Worktree");
        let entities = vec![entity(&repo, "repo", Kind::Repo)];
        let siblings = [sibling.clone()];

        let report = run(
            &plan(Operation::Delete, &entities),
            &config_file,
            |_| None,
            |_| siblings.to_vec(),
            |_| Vec::new(),
            |_| panic!("this test does not exercise sync"),
            |_| panic!("this test declares no before_sync hook"),
            |_| panic!("this test declares no after_sync hook"),
        );

        assert!(!sibling.exists(), "the cascade's own removal is a fact");
        assert!(
            repo.exists(),
            "and the Repo's own working tree is still on disk"
        );
        assert!(
            matches!(report.records[0].outcome, Outcome::Failed(_)),
            "the selected row itself failed, got {:?}",
            report.records[0].outcome
        );
        assert_eq!(
            report.removed_keys(),
            vec![EntityKey::new(Arc::from(sibling.as_path()))],
            "and the directory that did go is still what the run dismisses"
        );
    }

    /// The key a cascade reports is the resolved path, never git's own register entry
    /// verbatim: discovery keys every row by a resolved absolute directory, and a key that
    /// misses one is a row nothing drops.
    #[test]
    fn a_cascade_reports_the_resolved_key_for_a_worktree_git_records_unresolved() {
        let dir = tempfile::tempdir().expect("temp dir");
        let root = dir
            .path()
            .canonicalize()
            .expect("canonicalize the temp dir");
        let config_file = root.join("config.toml");
        let repo = root.join("repo");
        std::fs::create_dir_all(repo.join(".git")).expect("create the repo fixture");
        let sibling = root.join("sibling");
        std::fs::create_dir_all(sibling.join(".git")).expect("create the linked Worktree");
        // The shape git's own register hands back for an entry it recorded relative to the
        // Repo: the right directory, spelled a way no discovered key is.
        let as_recorded = repo.join("..").join("sibling");
        let entities = vec![entity(&repo, "repo", Kind::Repo)];

        let report = run(
            &plan(Operation::Delete, &entities),
            &config_file,
            |_| None,
            |_| vec![as_recorded.clone()],
            |_| Vec::new(),
            |_| panic!("this test does not exercise sync"),
            |_| panic!("this test declares no before_sync hook"),
            |_| panic!("this test declares no after_sync hook"),
        );

        assert!(!sibling.exists(), "the linked Worktree is gone");
        assert_eq!(
            report.removed_keys(),
            vec![
                EntityKey::new(Arc::from(sibling.as_path())),
                entities[0].key.clone(),
            ],
            "each key names the directory discovery would have keyed the row by"
        );
    }

    // =====================================================================================
    // `delete`'s phase 1 and phase 2: enumerating and draining a working tree's own ignored
    // directories before phase 3, `remove_working_tree`, runs (docs/spec/repo-management.md's
    // "Deleting a working tree").
    // =====================================================================================

    /// [`delete_ignored_directories`] itself, independent of `delete_one`: every directory it
    /// is given disappears, nested content included.
    #[test]
    fn delete_ignored_directories_removes_each_directory_it_is_given() {
        let dir = tempfile::tempdir().expect("temp dir");
        let node_modules = dir.path().join("node_modules");
        std::fs::create_dir_all(node_modules.join("a-package")).expect("create node_modules");
        std::fs::write(node_modules.join("a-package").join("index.js"), "x")
            .expect("write nested file");
        let target = dir.path().join("target");
        std::fs::create_dir_all(&target).expect("create target");

        let failed = delete_ignored_directories(vec![node_modules.clone(), target.clone()]);

        assert!(
            failed.is_empty(),
            "both directories should remove cleanly, got {failed:?}"
        );
        assert!(!node_modules.exists());
        assert!(!target.exists());
    }

    /// A directory that will not remove is reported rather than panicking, and does not stop
    /// a sibling directory from being drained: `delete`'s own phase 3 still runs afterwards
    /// regardless of what phase 2 could not finish.
    #[test]
    fn delete_ignored_directories_reports_a_directory_that_will_not_remove_without_stopping_the_rest()
     {
        let dir = tempfile::tempdir().expect("temp dir");
        let missing = dir.path().join("already-gone");
        let present = dir.path().join("present");
        std::fs::create_dir_all(&present).expect("create present");

        let failed = delete_ignored_directories(vec![missing.clone(), present.clone()]);

        assert_eq!(failed, vec![missing]);
        assert!(
            !present.exists(),
            "a failure removing one directory must not hold back the rest"
        );
    }

    #[test]
    fn delete_ignored_directories_does_nothing_when_given_nothing() {
        assert_eq!(
            delete_ignored_directories(Vec::new()),
            Vec::<PathBuf>::new()
        );
    }

    /// The safety argument's other half: phase 2 is asked about, and only about, the exact
    /// path `delete_one` was given, a Worktree's own [`EntityKey`] path here, never a
    /// literal or a path phase 2 invented on its own.
    #[test]
    fn deleting_a_worktree_asks_for_ignored_directories_at_the_worktrees_own_path() {
        let dir = tempfile::tempdir().expect("temp dir");
        let config_file = dir.path().join("config.toml");
        let tree = dir.path().join("tree");
        std::fs::create_dir_all(tree.join(".git")).expect("create the worktree fixture");
        let entities = vec![entity(&tree, "tree", Kind::Worktree)];
        let asked = std::cell::RefCell::new(Vec::new());

        let report = run(
            &plan(Operation::Delete, &entities),
            &config_file,
            |_| None,
            |_| Vec::new(),
            |path| {
                asked.borrow_mut().push(path.to_path_buf());
                Vec::new()
            },
            |_| panic!("this test does not exercise sync"),
            |_| panic!("this test declares no before_sync hook"),
            |_| panic!("this test declares no after_sync hook"),
        );

        assert_eq!(asked.into_inner(), vec![tree.clone()]);
        assert_eq!(
            report.records[0].outcome,
            removed(Removal::Directory, ConfigCleanup::NoEntryOfItsOwn)
        );
    }

    /// The Repo cascade gets the same treatment as the Repo's own tree: phase 1 and phase 2
    /// run once for the Repo's own path and once more for each linked Worktree
    /// `linked_worktree_paths` names, never only for one or the other.
    #[test]
    fn deleting_a_repo_asks_for_ignored_directories_at_its_own_path_and_every_linked_worktrees() {
        let dir = tempfile::tempdir().expect("temp dir");
        let config_file = dir.path().join("config.toml");
        let repo = dir.path().join("repo");
        std::fs::create_dir_all(repo.join(".git")).expect("create the repo fixture");
        let sibling = dir.path().join("sibling");
        std::fs::create_dir_all(sibling.join(".git")).expect("create the sibling");
        let entities = vec![entity(&repo, "repo", Kind::Repo)];
        let siblings = [sibling.clone()];
        let asked = std::cell::RefCell::new(Vec::new());

        let report = run(
            &plan(Operation::Delete, &entities),
            &config_file,
            |_| None,
            |_| siblings.to_vec(),
            |path| {
                asked.borrow_mut().push(path.to_path_buf());
                Vec::new()
            },
            |_| panic!("this test does not exercise sync"),
            |_| panic!("this test declares no before_sync hook"),
            |_| panic!("this test declares no after_sync hook"),
        );

        let mut asked = asked.into_inner();
        asked.sort();
        let mut expected = vec![repo.clone(), sibling.clone()];
        expected.sort();
        assert_eq!(asked, expected);
        assert_eq!(
            report.records[0].outcome,
            removed(Removal::WorkingTree, ConfigCleanup::NoEntryOfItsOwn)
        );
    }

    /// Phase 2 and phase 3 together remove exactly what the old single-phase
    /// `remove_working_tree` alone removed: a real ignored subtree drained by phase 2 leaves
    /// nothing behind for phase 3 to trip over, and the working tree ends up gone either way.
    #[test]
    fn phase_two_and_phase_three_together_remove_the_whole_working_tree() {
        let dir = tempfile::tempdir().expect("temp dir");
        let config_file = dir.path().join("config.toml");
        let tree = dir.path().join("tree");
        std::fs::create_dir_all(tree.join(".git")).expect("create the worktree fixture");
        let node_modules = tree.join("node_modules");
        std::fs::create_dir_all(node_modules.join("a-package")).expect("create node_modules");
        std::fs::write(node_modules.join("a-package").join("index.js"), "x")
            .expect("write nested file");
        std::fs::write(tree.join("source.rs"), "fn main() {}\n").expect("write a tracked file");
        let entities = vec![entity(&tree, "tree", Kind::Worktree)];

        let report = run(
            &plan(Operation::Delete, &entities),
            &config_file,
            |_| None,
            |_| Vec::new(),
            |path| vec![path.join("node_modules")],
            |_| panic!("this test does not exercise sync"),
            |_| panic!("this test declares no before_sync hook"),
            |_| panic!("this test declares no after_sync hook"),
        );

        assert!(
            !tree.exists(),
            "phase 2 draining node_modules must not stop phase 3 from removing the rest"
        );
        assert_eq!(
            report.records[0].outcome,
            removed(Removal::Directory, ConfigCleanup::NoEntryOfItsOwn)
        );
    }

    /// Crash recovery: phase 2 finishing (or failing) on its own, with `delete` itself never
    /// reaching phase 3, leaves the working tree still on disk and still eligible. Re-running
    /// `delete` finds less for phase 1 to enumerate and still finishes the removal phase 3
    /// would have finished the first time.
    #[test]
    fn a_working_tree_left_behind_after_phase_two_alone_still_deletes_cleanly_on_a_re_run() {
        let dir = tempfile::tempdir().expect("temp dir");
        let config_file = dir.path().join("config.toml");
        let tree = dir.path().join("tree");
        std::fs::create_dir_all(tree.join(".git")).expect("create the worktree fixture");
        let node_modules = tree.join("node_modules");
        std::fs::create_dir_all(node_modules.join("a-package")).expect("create node_modules");

        // Stands in for a crash right after phase 2 finished but before `delete` reached
        // phase 3 at all: the ignored subtree is already gone, the tree is still there.
        let failed = delete_ignored_directories(vec![node_modules.clone()]);
        assert!(failed.is_empty());
        assert!(!node_modules.exists());
        assert!(
            tree.exists(),
            "the crash this stands in for happens before phase 3 ever runs"
        );

        let entities = vec![entity(&tree, "tree", Kind::Worktree)];
        let report = run(
            &plan(Operation::Delete, &entities),
            &config_file,
            |_| None,
            |_| Vec::new(),
            |_| Vec::new(),
            |_| panic!("this test does not exercise sync"),
            |_| panic!("this test declares no before_sync hook"),
            |_| panic!("this test declares no after_sync hook"),
        );

        assert!(
            !tree.exists(),
            "the re-run finishes the removal the interrupted first run left half done"
        );
        assert_eq!(
            report.records[0].outcome,
            removed(Removal::Directory, ConfigCleanup::NoEntryOfItsOwn)
        );
    }

    // =====================================================================================
    // "One removal, reported once": a Worktree selected alongside the parent Repo it is
    // linked from is dropped from the Plan entirely, since the Repo's own delete already
    // takes it with it.
    // =====================================================================================

    #[test]
    fn a_worktree_selected_alongside_its_parent_repo_is_not_named_as_its_own_target() {
        let repo = entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo);
        let tree = worktree_of(&repo, Path::new("/tmp/x/tree"), "tree");
        let entities = vec![repo, tree];

        let plan = plan(Operation::Delete, &entities);

        assert_eq!(
            plan.targets.len(),
            1,
            "the Worktree covered by its selected parent must not be its own target"
        );
        assert_eq!(plan.targets[0].name.as_ref(), "repo");
        assert_eq!(plan.eligible_count(), 1);
    }

    /// A Worktree whose parent is not itself selected keeps its own target: the merge is
    /// about what is in the same gesture, not about family membership on its own.
    #[test]
    fn a_worktree_whose_parent_is_not_selected_keeps_its_own_target() {
        let repo = entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo);
        let tree = worktree_of(&repo, Path::new("/tmp/x/tree"), "tree");
        let entities = vec![tree.clone()];

        let plan = Plan::new(Operation::Delete, &entities, checked(&entities));

        assert_eq!(plan.targets.len(), 1);
        assert_eq!(plan.targets[0].name.as_ref(), "tree");
    }

    // =====================================================================================
    // Criterion 6: the confirm gate's three risk lines, and the Repo listed plainly.
    // =====================================================================================

    fn delete_plan_with(risk: DeleteRisk) -> Plan {
        let entities = vec![entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo)];
        plan(Operation::Delete, &entities).with_risk(|_| Ok(risk))
    }

    #[test]
    fn a_repo_with_none_of_the_three_risks_is_listed_plainly() {
        let plan = delete_plan_with(DeleteRisk {
            uncommitted: false,
            unpushed_commits: 0,
            unpushed_branches: 0,
            linked_worktrees: 0,
        });

        let lines = plan.confirm_lines();

        assert_eq!(
            lines[1], "repo",
            "a Repo with nothing to lose is its name and nothing else, got {lines:?}"
        );
    }

    #[test]
    fn each_risk_line_appears_only_when_its_own_fact_is_true() {
        let none = DeleteRisk {
            uncommitted: false,
            unpushed_commits: 0,
            unpushed_branches: 0,
            linked_worktrees: 0,
        };

        let uncommitted = delete_plan_with(DeleteRisk {
            uncommitted: true,
            ..none
        })
        .confirm_lines()[1]
            .clone();
        assert_eq!(uncommitted, "repo: uncommitted changes");

        let unpushed = delete_plan_with(DeleteRisk {
            unpushed_commits: 3,
            unpushed_branches: 2,
            ..none
        })
        .confirm_lines()[1]
            .clone();
        assert_eq!(unpushed, "repo: 3 commits unpushed on 2 branches");

        let worktrees = delete_plan_with(DeleteRisk {
            linked_worktrees: 1,
            ..none
        })
        .confirm_lines()[1]
            .clone();
        assert_eq!(worktrees, "repo: 1 linked worktree");

        let all_three = delete_plan_with(DeleteRisk {
            uncommitted: true,
            unpushed_commits: 1,
            unpushed_branches: 1,
            linked_worktrees: 2,
        })
        .confirm_lines()[1]
            .clone();
        assert_eq!(
            all_three,
            "repo: uncommitted changes, 1 commit unpushed on 1 branch, 2 linked worktrees"
        );
    }

    /// A Worktree row's own gate line never names a linked-Worktree count: deleting one
    /// Worktree never touches its siblings, so the count would mislead rather than inform.
    #[test]
    fn a_worktrees_own_gate_line_never_names_a_linked_worktree_count() {
        let tree = entity(Path::new("/tmp/x/tree"), "tree", Kind::Worktree);
        let plan = plan(Operation::Delete, &[tree]).with_risk(|_| {
            Ok(DeleteRisk {
                uncommitted: true,
                unpushed_commits: 0,
                unpushed_branches: 0,
                linked_worktrees: 3,
            })
        });

        let line = plan.confirm_lines()[1].clone();

        assert_eq!(
            line, "tree: uncommitted changes",
            "a Worktree's own family size is not this row's own risk, got {line:?}"
        );
    }

    /// A risk that would not read is said so, never zeroed: a gate that reported "nothing to
    /// lose" because it could not look is the worst answer available here.
    #[test]
    fn a_risk_that_could_not_be_read_is_said_rather_than_reported_as_nothing() {
        let entities = vec![entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo)];
        let plan =
            plan(Operation::Delete, &entities).with_risk(|_| Err("the refs would not list".into()));

        let line = plan.confirm_lines()[1].clone();

        assert!(
            line.contains("could not be read") && line.contains("the refs would not list"),
            "got {line:?}"
        );
    }

    /// The sentence itself is repo-management.md's, read at test time rather than restated
    /// here: the constant may be reworded, but never away from the document that requires it
    /// ("There is no undo and no trash, which the gate says in as many words").
    #[test]
    fn the_no_undo_sentence_is_repo_management_mds_own_words() {
        let spec = spec_source();
        let sentence = spec
            .split("A Repo with none of the three is listed plainly. ")
            .nth(1)
            .and_then(|rest| rest.split(", which the gate says in as many words").next())
            .expect("repo-management.md still names the sentence the gate must say");
        let mut characters = sentence.chars();
        let lowercased = match characters.next() {
            Some(first) => first.to_lowercase().to_string() + characters.as_str(),
            None => String::new(),
        };

        assert_eq!(
            NO_UNDO, lowercased,
            "the constant must be the specification's own sentence"
        );
    }

    #[test]
    fn the_delete_gate_says_there_is_no_undo_and_ignore_adds_no_lines_at_all() {
        let entities = vec![entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo)];

        let deleting = plan(Operation::Delete, &entities).confirm_lines();
        assert_eq!(
            deleting.last().map(String::as_str),
            Some(NO_UNDO),
            "the gate has to say it in as many words, got {deleting:?}"
        );

        let ignoring = plan(Operation::Ignore, &entities).confirm_lines();
        assert_eq!(
            ignoring,
            vec!["ignore on 1 selected?".to_string(), "repo".to_string()],
            "neither destroys anything, so neither gets an additional line"
        );
    }

    // =====================================================================================
    // The eligible column: `ignore` reaches an excluded row and a listed one alike, since
    // the run reads the row's own state to decide which direction it goes.
    // =====================================================================================

    #[test]
    fn ignore_is_eligible_on_an_excluded_row_and_a_listed_one_alike() {
        let plain = entity(Path::new("/tmp/x/a"), "a", Kind::Repo);
        let already = excluded(entity(Path::new("/tmp/x/b"), "b", Kind::Repo));

        assert_eq!(Operation::Ignore.eligibility(&plain), Eligibility::Eligible);
        assert_eq!(
            Operation::Ignore.eligibility(&already),
            Eligibility::Eligible
        );
    }

    #[test]
    fn a_worktree_is_eligible_to_ignore_and_a_submodule_is_not() {
        let worktree = entity(Path::new("/tmp/x/tree"), "tree", Kind::Worktree);
        let submodule = entity(Path::new("/tmp/x/sub"), "sub", Kind::Submodule);

        assert_eq!(
            Operation::Ignore.eligibility(&worktree),
            Eligibility::Eligible
        );
        assert_eq!(
            Operation::Ignore.eligibility(&submodule),
            Eligibility::Refused(Refusal::SubmoduleHasNoEntryOfItsOwn)
        );
    }

    // =====================================================================================
    // `sync`'s own eligibility: a Repo is always eligible by Kind; a Worktree and a
    // Submodule are always refused with a reason of their own.
    // =====================================================================================

    /// A Repo is eligible for `sync`.
    #[test]
    fn sync_is_eligible_on_a_repo() {
        let repo = entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo);

        let eligibility = Operation::Sync.eligibility(&repo);

        assert_eq!(eligibility, Eligibility::Eligible);
    }

    /// A Worktree is refused with its own reason, never silently ineligible:
    /// `repos_eligible_for_auto_update_attempt` is Repo-only, so a Worktree sharing a common
    /// dir must say so rather than doing nothing.
    #[test]
    fn sync_is_refused_on_a_worktree_and_named_and_counted() {
        let worktree = entity(Path::new("/tmp/x/tree"), "tree", Kind::Worktree);

        let eligibility = Operation::Sync.eligibility(&worktree);

        assert_eq!(
            eligibility,
            Eligibility::Refused(Refusal::WorktreeSyncsThroughItsRepo)
        );

        let entities = vec![worktree];
        let plan = plan(Operation::Sync, &entities);
        assert_eq!(
            plan.eligible_count(),
            0,
            "a Worktree is never eligible for sync"
        );
        assert_eq!(plan.refused_count(), 1, "and the refusal is counted");
    }

    /// A Submodule is refused with its own reason: it tracks a pinned commit, not a branch.
    #[test]
    fn sync_is_refused_on_a_submodule() {
        let submodule = entity(Path::new("/tmp/x/sub"), "sub", Kind::Submodule);

        let eligibility = Operation::Sync.eligibility(&submodule);

        assert_eq!(
            eligibility,
            Eligibility::Refused(Refusal::SubmoduleCannotSync)
        );
    }

    // =====================================================================================
    // `sync`'s own run: every `AutoUpdateAttempt` the injected closure returns becomes the
    // matching `Outcome`, since the closure stands in for `Core::attempt_auto_update` here,
    // the same way `with_risk`'s own `read` stands in for `Core::delete_risk`.
    // =====================================================================================

    fn run_with_sync(plan: &Plan, attempt: AutoUpdateAttempt) -> Report {
        run(
            plan,
            Path::new("/tmp/unused-config.toml"),
            |_| None,
            |_| Vec::new(),
            |_| Vec::new(),
            move |_| attempt.clone(),
            |_| None,
            |_| None,
        )
    }

    /// [`run_with_sync`] plus a `before_sync` and an `after_sync` hook of the caller's own
    /// choosing, for the hook-specific tests below.
    fn run_with_sync_and_hooks(
        plan: &Plan,
        attempt: AutoUpdateAttempt,
        before_sync: Option<HookOutcome>,
        after_sync: Option<HookOutcome>,
    ) -> Report {
        run(
            plan,
            Path::new("/tmp/unused-config.toml"),
            |_| None,
            |_| Vec::new(),
            |_| Vec::new(),
            move |_| attempt.clone(),
            move |_| before_sync.clone(),
            move |_| after_sync.clone(),
        )
    }

    #[test]
    fn sync_updated_becomes_synced() {
        let entities = vec![entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo)];
        let built = plan(Operation::Sync, &entities);

        let report = run_with_sync(&built, AutoUpdateAttempt::Updated);

        assert_eq!(report.records[0].outcome, Outcome::Synced);
        assert!(
            report.summary().contains("1 done"),
            "got {:?}",
            report.summary()
        );
    }

    #[test]
    fn every_auto_update_ineligible_reason_reaches_the_report_as_a_reason() {
        let entities = vec![entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo)];
        let cases = [
            (AutoUpdateAttempt::NotClean, SyncIneligibility::NotClean),
            (AutoUpdateAttempt::NoUpstream, SyncIneligibility::NoUpstream),
            (AutoUpdateAttempt::NotBehind, SyncIneligibility::NotBehind),
            (
                AutoUpdateAttempt::NotFastForward,
                SyncIneligibility::NotFastForward,
            ),
        ];

        for (attempt, expected) in cases {
            let built = plan(Operation::Sync, &entities);

            let report = run_with_sync(&built, attempt.clone());

            assert_eq!(
                report.records[0].outcome,
                Outcome::NotEligibleToSync(expected),
                "attempt {attempt:?} must surface as a reason, never silently"
            );
            assert!(
                own_work(&report.records[0].outcome)
                    .said()
                    .contains(expected.reason()),
                "the receipt's own words must carry the reason"
            );
            assert!(
                report.summary().contains("1 not eligible to sync"),
                "got {:?}",
                report.summary()
            );
        }
    }

    #[test]
    fn sync_failed_becomes_a_failure_never_an_ineligible_reason() {
        let entities = vec![entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo)];
        let built = plan(Operation::Sync, &entities);

        let report = run_with_sync(&built, AutoUpdateAttempt::Failed("git said no".to_string()));

        assert!(
            matches!(&report.records[0].outcome, Outcome::Failed(message) if message == "git said no")
        );
    }

    // =====================================================================================
    // Hooks around sync (docs/spec/repo-management.md's "Hooks around sync"): a pre hook
    // that fails stops sync from being attempted at all; a post hook that fails never undoes
    // a fast-forward that already happened.
    // =====================================================================================

    /// A pre hook that fails must stop `sync` before it is ever attempted: `attempt_sync`
    /// itself panics if called, so this only passes if `sync_one` never reaches it.
    #[test]
    fn a_failing_before_sync_hook_stops_sync_from_being_attempted() {
        let entities = vec![entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo)];
        let built = plan(Operation::Sync, &entities);

        let report = run(
            &built,
            Path::new("/tmp/unused-config.toml"),
            |_| None,
            |_| Vec::new(),
            |_| Vec::new(),
            |_| panic!("a failing pre-hook must stop sync before attempt_sync is ever called"),
            |_| Some(HookOutcome::Failed("exit 1".to_string())),
            |_| panic!("a before_sync failure must never reach the after_sync hook either"),
        );

        assert_eq!(
            report.records[0].outcome,
            Outcome::BeforeSyncHookFailed("exit 1".to_string())
        );
        assert!(
            matches!(
                own_work(&report.records[0].outcome),
                OwnWork::CouldNotAct(_)
            ),
            "sync never ran, so this row could not act rather than merely being refused"
        );
    }

    /// A passing pre hook lets `sync` proceed exactly as it would with none declared.
    #[test]
    fn a_passing_before_sync_hook_still_lets_sync_run() {
        let entities = vec![entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo)];
        let built = plan(Operation::Sync, &entities);

        let report = run_with_sync_and_hooks(
            &built,
            AutoUpdateAttempt::Updated,
            Some(HookOutcome::Passed),
            None,
        );

        assert_eq!(report.records[0].outcome, Outcome::Synced);
    }

    /// A post hook that fails never undoes the fast-forward: the row still reports the
    /// branch moved, with the hook's own failure carried alongside it rather than replacing
    /// it, so "sync happened" is never lost.
    #[test]
    fn a_failing_after_sync_hook_never_undoes_the_fast_forward_it_already_did() {
        let entities = vec![entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo)];
        let built = plan(Operation::Sync, &entities);

        let report = run_with_sync_and_hooks(
            &built,
            AutoUpdateAttempt::Updated,
            None,
            Some(HookOutcome::Failed("exit 1".to_string())),
        );

        assert_eq!(
            report.records[0].outcome,
            Outcome::SyncedAfterHookFailed("exit 1".to_string())
        );
        assert!(
            matches!(own_work(&report.records[0].outcome), OwnWork::Did(message) if message.contains("fast-forwarded")),
            "the fast-forward already happened and must still read as done, got {:?}",
            own_work(&report.records[0].outcome)
        );
        assert!(
            report.summary().contains("1 after_sync hook failed"),
            "the summary must flag the hook failure rather than folding it silently into \
             'done', got {:?}",
            report.summary()
        );
    }

    /// The after_sync hook is never even consulted when `sync` did not actually fast-forward
    /// the branch: nothing happened for it to run after.
    #[test]
    fn after_sync_hook_is_not_consulted_when_sync_did_not_fast_forward() {
        let entities = vec![entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo)];
        let built = plan(Operation::Sync, &entities);

        let report = run(
            &built,
            Path::new("/tmp/unused-config.toml"),
            |_| None,
            |_| Vec::new(),
            |_| Vec::new(),
            |_| AutoUpdateAttempt::NotBehind,
            |_| None,
            |_| panic!("after_sync must never be consulted when sync did not fast-forward"),
        );

        assert_eq!(
            report.records[0].outcome,
            Outcome::NotEligibleToSync(SyncIneligibility::NotBehind)
        );
    }

    /// A passing after_sync hook reports the plain `Synced` outcome, with no trace that a
    /// hook ran at all: only a hook's own failure is worth a distinct outcome.
    #[test]
    fn a_passing_after_sync_hook_leaves_the_outcome_as_plain_synced() {
        let entities = vec![entity(Path::new("/tmp/x/repo"), "repo", Kind::Repo)];
        let built = plan(Operation::Sync, &entities);

        let report = run_with_sync_and_hooks(
            &built,
            AutoUpdateAttempt::Updated,
            None,
            Some(HookOutcome::Passed),
        );

        assert_eq!(report.records[0].outcome, Outcome::Synced);
    }

    /// [`hook_outcome_from_receipt`]'s own job: a receipt with a failing step becomes
    /// `Failed` naming that step, and a receipt whose steps all ran clean becomes `Passed`.
    #[test]
    fn hook_outcome_from_receipt_reads_the_first_failing_step() {
        use repon_core::{ActionReceipt, StepOutcome, StepResult};
        use std::sync::Arc as StdArc;
        use std::time::Duration;

        let passing = ActionReceipt {
            label: StdArc::from("hook"),
            steps: StdArc::from(vec![StepResult {
                label: StdArc::from("true"),
                outcome: StepOutcome::Ok,
                output: StdArc::from(&b""[..]),
                elapsed: Duration::ZERO,
                elision: None,
                shell: false,
                interactive: false,
            }]),
            skip: None,
            finished_at: repon_core::Timestamp::now(),
            running: None,
        };
        assert_eq!(hook_outcome_from_receipt(&passing), HookOutcome::Passed);

        let failing = ActionReceipt {
            steps: StdArc::from(vec![StepResult {
                label: StdArc::from("false"),
                outcome: StepOutcome::Failed(1),
                output: StdArc::from(&b""[..]),
                elapsed: Duration::ZERO,
                elision: None,
                shell: false,
                interactive: false,
            }]),
            ..passing
        };
        assert_eq!(
            hook_outcome_from_receipt(&failing),
            HookOutcome::Failed("`false` exited 1".to_string())
        );
    }

    /// `ignore` covers both directions from one palette entry: run over a row an entry
    /// already excludes, it removes what the first run wrote, so a config file that had no
    /// `[[repo]]` array is byte for byte what it was.
    #[test]
    fn running_ignore_twice_returns_the_config_file_byte_for_byte() {
        let dir = tempfile::tempdir().expect("temp dir");
        let config_file = dir.path().join("config.toml");
        let before = "# a comment worth keeping\ntheme = \"default\"\n";
        std::fs::write(&config_file, before).expect("write the config file");
        let plain = entity(&dir.path().join("repo"), "repo", Kind::Repo);

        run_plain(
            &plan(Operation::Ignore, std::slice::from_ref(&plain)),
            &config_file,
        );
        let ignored = std::fs::read_to_string(&config_file).expect("read it back");
        assert!(ignored.contains("exclude = true"), "got {ignored:?}");

        run_plain(&plan(Operation::Ignore, &[excluded(plain)]), &config_file);

        assert_eq!(
            std::fs::read_to_string(&config_file).expect("read it back"),
            before,
            "the second `ignore` removes the key the first one wrote"
        );
    }

    /// A Worktree excluded through the entry naming its Repo is not silently reported as
    /// no longer ignored: removing that entry would show every entity sharing the git common
    /// dir again, so nothing is written and the row says which it is.
    #[test]
    fn ignore_on_a_row_excluded_by_an_inherited_entry_writes_nothing_and_says_so() {
        let dir = tempfile::tempdir().expect("temp dir");
        let config_file = dir.path().join("config.toml");
        let before = "[[repo]]\npath = \"/somewhere/else\"\nexclude = true\n";
        std::fs::write(&config_file, before).expect("write the config file");
        let inheriting = excluded(entity(&dir.path().join("tree"), "tree", Kind::Worktree));

        let report = run_plain(&plan(Operation::Ignore, &[inheriting]), &config_file);

        assert_eq!(
            report.records[0].outcome,
            Outcome::ExcludedByAnInheritedEntry
        );
        assert_eq!(
            std::fs::read_to_string(&config_file).expect("read it back"),
            before,
            "the entry naming another path must be left alone"
        );
        assert!(
            report.summary().contains("still ignored by another entry"),
            "got {:?}",
            report.summary()
        );
    }

    /// The two guards on the one call that destroys work: neither can act, both can only
    /// refuse, and each is reported as a failure rather than passing silently.
    #[test]
    fn deleting_refuses_a_relative_path_and_a_directory_that_is_not_a_repository() {
        let relative = remove_working_tree(Path::new("relative/repo"))
            .expect_err("a relative path must never be removed");
        assert!(relative.to_string().contains("relative"));

        let dir = tempfile::tempdir().expect("temp dir");
        let not_a_repo = dir.path().join("plain-directory");
        std::fs::create_dir_all(&not_a_repo).expect("create it");

        let refused = remove_working_tree(&not_a_repo)
            .expect_err("a directory with no `.git` must never be removed");

        assert!(refused.to_string().contains(".git"));
        assert!(not_a_repo.exists(), "and it is still there");
    }

    // =====================================================================================
    // The receipt: a management run's own result, docs/spec/repo-management.md's "Receipts".
    // =====================================================================================

    /// Every [`Outcome`] earns a grade of own work, and no grade is decided by hand at a
    /// second site: `describe`, the log line's own words, reads them out of [`own_work`]. The
    /// pairing is the receipts table in repo-management.md, read at test time rather than
    /// restated, so a sentence that drifted from the specification fails here.
    #[test]
    fn every_outcome_maps_to_a_grade_of_own_work_whose_words_the_spec_carries() {
        let spec = spec_source();
        let receipts = spec
            .split("## Receipts")
            .nth(1)
            .expect("repo-management.md still carries a Receipts section");

        let cases = [
            (Outcome::Ignored, "Did"),
            (Outcome::Unignored, "Did"),
            (
                removed(Removal::WorkingTree, ConfigCleanup::EntryRemoved),
                "Did",
            ),
            (
                removed(Removal::WorkingTree, ConfigCleanup::NoEntryOfItsOwn),
                "Did",
            ),
            (
                removed(Removal::Worktree, ConfigCleanup::EntryRemoved),
                "Did",
            ),
            (
                removed(Removal::Worktree, ConfigCleanup::NoEntryOfItsOwn),
                "Did",
            ),
            (
                removed(Removal::Directory, ConfigCleanup::EntryRemoved),
                "Did",
            ),
            (
                removed(Removal::Directory, ConfigCleanup::NoEntryOfItsOwn),
                "Did",
            ),
            (Outcome::ExcludedByAnInheritedEntry, "Refused"),
            (
                Outcome::Refused(Refusal::SubmoduleHasNoEntryOfItsOwn),
                "Refused",
            ),
            (Outcome::Failed("boom".to_string()), "CouldNotAct"),
            (Outcome::Synced, "Did"),
            (
                Outcome::BeforeSyncHookFailed("boom".to_string()),
                "CouldNotAct",
            ),
            (Outcome::SyncedAfterHookFailed("boom".to_string()), "Did"),
        ];

        for (outcome, grade) in cases {
            let work = own_work(&outcome);
            let named = match &work {
                OwnWork::Did(_) => "Did",
                OwnWork::Refused(_) => "Refused",
                OwnWork::CouldNotAct(_) => "CouldNotAct",
            };
            assert_eq!(named, grade, "{outcome:?} took the wrong grade");
            assert_eq!(
                describe(&outcome),
                work.said().to_string(),
                "the log line and the receipt must read the same words for {outcome:?}"
            );
            assert!(
                receipts.contains(&format!("`{grade}`")),
                "repo-management.md's Receipts section no longer names the `{grade}` grade"
            );
        }
    }

    /// The four words the specification's own Receipts table gives the rows nothing else in
    /// this module composes, read out of the document rather than restated beside it: a
    /// sentence changed in one place and not the other fails here.
    #[test]
    fn the_receipts_own_words_are_repo_management_mds_own() {
        let spec = spec_source();
        let receipts = spec
            .split("## Receipts")
            .nth(1)
            .expect("repo-management.md still carries a Receipts section");

        for outcome in [
            Outcome::Ignored,
            Outcome::Unignored,
            removed(Removal::WorkingTree, ConfigCleanup::EntryRemoved),
            removed(Removal::WorkingTree, ConfigCleanup::NoEntryOfItsOwn),
            removed(Removal::Worktree, ConfigCleanup::EntryRemoved),
            removed(Removal::Worktree, ConfigCleanup::NoEntryOfItsOwn),
            removed(Removal::Directory, ConfigCleanup::EntryRemoved),
            removed(Removal::Directory, ConfigCleanup::NoEntryOfItsOwn),
            Outcome::ExcludedByAnInheritedEntry,
        ] {
            let said = describe(&outcome);
            assert!(
                receipts.contains(&said),
                "repo-management.md's Receipts table does not carry {said:?}"
            );
        }
    }

    /// A run's records reach [`repon_core::Core::record_own_work`] whole: every Selection row,
    /// refusals included, each carrying the Entity it names so a receipt cannot land on the
    /// wrong row.
    #[test]
    fn every_row_of_a_run_including_a_refusal_becomes_an_own_work_record_for_its_own_entity() {
        let dir = tempfile::tempdir().expect("temp dir");
        let config_file = dir.path().join("config.toml");
        let repo = dir.path().join("repo");
        let sub = dir.path().join("sub");
        let entities = vec![
            entity(&repo, "repo", Kind::Repo),
            entity(&sub, "sub", Kind::Submodule),
        ];

        let report = run_plain(&plan(Operation::Ignore, &entities), &config_file);
        let records = report.own_work_records();

        assert_eq!(records.len(), 2, "every row is recorded, refusals included");
        assert_eq!(
            records[0].0, entities[0].key,
            "in the Selection's own order"
        );
        assert_eq!(records[1].0, entities[1].key);
        assert!(matches!(records[0].1, OwnWork::Did(_)));
        assert!(
            matches!(&records[1].1, OwnWork::Refused(said) if said.contains("Submodule")),
            "the refused row carries the gate's own reason, got {:?}",
            records[1].1
        );
    }

    /// The register entry this replaced is gone, and the document that owns the answer records
    /// it rather than the gap: the same shape
    /// `actions_md_records_the_settled_answer_for_per_repo_applicability` holds for its own
    /// entry, so a register that kept the entry after the answer landed fails here.
    #[test]
    fn repo_management_md_records_the_receipt_and_the_register_no_longer_carries_the_gap() {
        let spec = spec_source();
        assert!(
            !spec.contains("Not built."),
            "repo-management.md still records the receipt as not built"
        );
        assert!(
            spec.contains("## Receipts"),
            "repo-management.md must still own the Receipts section"
        );

        let manifest_dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR"));
        let register = std::fs::read_to_string(manifest_dir.join("../../docs/open-questions.md"))
            .expect("read docs/open-questions.md");
        assert!(
            !register.contains("## A management result has no receipt of its own"),
            "the register keeps an entry its owning document has now answered"
        );

        let actions = std::fs::read_to_string(manifest_dir.join("../../docs/spec/actions.md"))
            .expect("read docs/spec/actions.md");
        assert!(
            actions.contains("A closed set of five."),
            "actions.md owns the outcome set and must declare the fifth"
        );
        assert!(
            actions.contains("Why the set grew from four to five"),
            "actions.md must say why the set grew, not only that it did"
        );
    }

    /// Every arm of every `match` over a Step outcome across both crates, and none of them a
    /// catch-all: `docs/spec/actions.md` calls the set closed and the compiler only enforces
    /// that where no `_` arm swallows what it has not been taught. `is_failure`'s own doc
    /// comment makes that promise for one match; this is the promise for the rest.
    ///
    /// Both crates' `src`, through [`crate::test_support::workspace_crate_src_dirs`]: the
    /// outcome is defined in `repon-core` and rendered in this crate, so either half alone is
    /// a scan that has stopped scanning. A match whose own arms never name the outcome is not
    /// this claim's subject and is skipped; an arm of a nested match is at a deeper
    /// indentation than its parent's and belongs to the nested one.
    #[test]
    fn no_match_over_a_step_outcome_anywhere_in_either_crate_has_a_catch_all_arm() {
        let mut offending = Vec::new();
        let mut matches_checked = 0usize;

        for dir in crate::test_support::workspace_crate_src_dirs() {
            for path in crate::test_support::rust_source_files(&dir) {
                let source = crate::test_support::production_source_at(&path);
                for arms in step_outcome_match_arms(&source) {
                    matches_checked += 1;
                    for arm in arms {
                        if is_catch_all(&arm) {
                            offending.push(format!("{}: {arm}", path.display()));
                        }
                    }
                }
            }
        }

        assert_eq!(
            matches_checked, 7,
            "the seven matches over a Step outcome this workspace holds are `is_failure`, \
             `is_refusal`, `OwnWork::said`, `step_outcome_word`, `step_outcome_meaning`, \
             `finished_step_line` and `describe_step_failure`; a different count means the \
             scan has stopped finding them, or an eighth landed and belongs on this list"
        );
        assert!(
            offending.is_empty(),
            "a match over a Step outcome reaches a fifth variant through a catch-all rather \
             than naming it: {offending:?}"
        );
    }

    /// The arm patterns of every `match` block in `source` whose own arms name a Step
    /// outcome, one `Vec` per block. Arms are the block lines at exactly one indent step
    /// inside the `match` line's own, which is what keeps a nested match's arms out.
    fn step_outcome_match_arms(source: &str) -> Vec<Vec<String>> {
        let lines: Vec<&str> = source.lines().collect();
        let mut blocks = Vec::new();
        for (index, line) in lines.iter().enumerate() {
            let trimmed = line.trim_start();
            if !trimmed.starts_with("match ") || !line.trim_end().ends_with('{') {
                continue;
            }
            let Some(block) = crate::test_support::block_at(source, index) else {
                continue;
            };
            let indent = line.len() - trimmed.len();
            let arm_indent = " ".repeat(indent + 4);
            let arms: Vec<String> = block
                .lines()
                .skip(1)
                .filter(|arm| {
                    arm.starts_with(&arm_indent) && !arm[arm_indent.len()..].starts_with(' ')
                })
                .map(|arm| match arm.split_once("=>") {
                    Some((pattern, _)) => pattern.trim().to_string(),
                    None => arm.trim().to_string(),
                })
                .collect();
            if arms
                .iter()
                .any(|arm| arm.contains("StepOutcome::") || arm.contains("OwnWork::"))
            {
                blocks.push(arms);
            }
        }
        blocks
    }

    /// Whether an arm pattern matches anything the arms above it did not, which is what makes
    /// a closed set stop being closed. A `_` alone, a `_` behind a guard, and a `_` as one
    /// alternative of a `|` pattern all count.
    fn is_catch_all(pattern: &str) -> bool {
        pattern
            .split('|')
            .map(str::trim)
            .any(|alternative| alternative == "_" || alternative.starts_with("_ if"))
    }
}