shepherd-cli 6.7.1

//! Native `shepherd run` lifecycle commands.
//!
//! The only writable namespace is `ExecutionContext::runs_root`, which is
//! resolved against the primary checkout.  Every state mutation goes through
//! `RunStore`, preserving its advisory lock and atomic canonical writer.

use std::{
    fs,
    io::Write,
    path::{Path, PathBuf},
    process::Command,
};

use sha2::{Digest, Sha256};
use shepherd::{
    RunState,
    digest::sha256_hex,
    run::{LaneState, LaneStatus, RunKind, RunStatus},
};

use crate::{
    ContextInputs, ExecutionContext, RunStore, RunStoreError,
    interface::{CliError, CliGlobals},
};

// Directories a run must have on disk. `lanes/` is the durable canonical
// artifact; reports/audits stay disposable views and are not scaffolded.
// `dispatch/` is operational rather than canonical, but it still has to exist
// before anything reads it: `DispatchStore` creates it on a write path only, so
// a run whose first dispatch touch is a read (resolving a root binding during
// PreToolUse) fails on a directory nothing had yet created. Scaffolding it here
// is what makes `run init` leave a namespace that can actually dispatch.
const RUN_SUBDIRS: [&str; 2] = ["lanes", "dispatch"];
const TRACKED_FILES: [&str; 6] = [
    "seed.md",
    "mesh.md",
    "plan.md",
    "phase0.md",
    "close.md",
    "handoff.md",
];
const LEDGER_FILE: &str = "auditor-verdicts.txt";
const SUCCESSOR_DOCUMENT_LIMIT: usize = 1024 * 1024;
const SUCCESSOR_ENTRY_LIMIT: usize = 4096;
const SUCCESSOR_DEPTH_LIMIT: usize = 16;
const SUCCESSOR_PATH_LIMIT: usize = 512;
const SUCCESSOR_FILE_LIMIT: u64 = 16 * 1024 * 1024;
const SUCCESSOR_TOTAL_LIMIT: u64 = 256 * 1024 * 1024;

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Args)]
#[command(disable_help_subcommand = true)]
pub struct RunCmd {
    #[command(subcommand)]
    action: Option<RunAction>,
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Subcommand)]
enum RunAction {
    Init {
        run: Option<String>,
        #[arg(long, default_value = "sprint")]
        kind: RunKind,
        #[arg(long, default_value = "")]
        branch: String,
        #[arg(long, default_value = "")]
        base: String,
        #[arg(long, default_value = "")]
        version: String,
        #[arg(long)]
        force: bool,
    },
    Rename {
        old: String,
        new: String,
    },
    Canonicalize {
        run: Option<String>,
        #[arg(long = "all")]
        all_runs: bool,
        #[arg(long)]
        dry_run: bool,
    },
    Show {
        run: String,
        #[arg(long)]
        json: bool,
    },
    List {
        #[arg(long)]
        json: bool,
    },
    Claim {
        run: String,
        #[arg(long)]
        json: bool,
    },
    Migrate {
        run: Option<String>,
        #[arg(long = "all")]
        all_runs: bool,
        /// Initialize Native identity for one pristine legacy planted run.
        #[arg(long)]
        adopt_native: bool,
    },
    /// Preserve one degraded incarnation and mint a fresh canonical
    /// incarnation for the same release version.
    Successor {
        run: String,
        #[arg(long)]
        source_incarnation: String,
        #[arg(long)]
        version: String,
        #[arg(long)]
        branch: String,
        #[arg(long)]
        base: String,
        #[arg(long)]
        source_baseline: String,
        #[arg(long)]
        baseline: String,
        #[arg(long)]
        worktree: PathBuf,
        #[arg(long)]
        confirm: bool,
        #[arg(long)]
        json: bool,
    },
    /// Abort one exact unactivated successor intent while preserving its
    /// source and failed-transition evidence.
    SuccessorAbort {
        run: String,
        #[arg(long)]
        source_incarnation: String,
        #[arg(long)]
        target_incarnation: String,
        #[arg(long)]
        confirm: bool,
        #[arg(long)]
        json: bool,
    },
    /// Move a run to a new lifecycle status.
    ///
    /// Separate from `set` because it is not a field assignment. `planned` and
    /// `executing` re-verify the run incarnation, both checkpoints, the source
    /// ledger, the Critic verdict, and the persisted post bytes, under the run
    /// lock. They were reachable through `set --status` only by an
    /// `if status ==` ladder, which is how a conformance case called `set-ok`
    /// came to expect exit 0 for a transition the engine refuses.
    Transition {
        run: String,
        #[arg(long)]
        to: shepherd::run::RunStatus,
    },
    /// Set a run's identity fields. Cannot express a status transition.
    Set {
        run: String,
        // `Option<T>`, not `default_value = ""`. An empty string is not a
        // value a caller can mean here, so using one as the absent marker made
        // "unset" and "set to empty" the same request and forced every read
        // through `is_empty()`.
        #[arg(long)]
        seed: Option<String>,
        #[arg(long)]
        plan: Option<String>,
        #[arg(long)]
        branch: Option<String>,
        #[arg(long)]
        base: Option<String>,
    },
    Orientation {
        #[command(subcommand)]
        action: OrientationAction,
    },
    Lane {
        #[command(subcommand)]
        action: LaneAction,
    },
    Wave {
        #[command(subcommand)]
        action: WaveAction,
    },
    Ledger {
        #[command(subcommand)]
        action: LedgerAction,
    },
    Layout {
        run: String,
        #[arg(long)]
        repair: bool,
        #[arg(long)]
        json: bool,
    },
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Subcommand)]
enum OrientationAction {
    Pre {
        run: String,
        #[arg(long)]
        json: bool,
    },
    Post {
        run: String,
        #[arg(long)]
        json: bool,
    },
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Subcommand)]
enum LaneAction {
    Add {
        run: String,
        lane: String,
        #[arg(long, default_value = "")]
        plan: String,
        #[arg(long, default_value = "")]
        worktree: String,
        #[arg(long, default_value = "")]
        branch: String,
    },
    Set {
        run: String,
        lane: String,
        #[arg(long)]
        state: LaneStatus,
    },
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Subcommand)]
enum WaveAction {
    Accept {
        run: String,
        lane: String,
        #[arg(long)]
        commit: String,
    },
    Merged {
        run: String,
        lane: String,
    },
    Pending {
        run: String,
        #[arg(long)]
        json: bool,
    },
    Verify {
        run: String,
        #[arg(long)]
        wave: Option<u32>,
        #[arg(long)]
        json: bool,
    },
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Subcommand)]
enum LedgerAction {
    Path {
        run: Option<String>,
        #[arg(long)]
        check: bool,
    },
    Check {
        run: Option<String>,
        #[arg(long)]
        json: bool,
    },
}

impl RunCmd {
    pub(crate) fn run(self, globals: CliGlobals) -> Result<(), CliError> {
        let mut context = context(globals)?;
        let Some(action) = self.action else {
            return output(&mut context, run_usage());
        };
        match action {
            RunAction::Init {
                run,
                kind,
                branch,
                base,
                version,
                force,
            } => init(&mut context, run, kind, &branch, &base, &version, force),
            RunAction::Rename { old, new } => rename(&mut context, &old, &new, true),
            RunAction::Canonicalize {
                run,
                all_runs,
                dry_run,
            } => canonicalize(&mut context, run, all_runs, dry_run),
            RunAction::Show { run, json } => show(&mut context, &run, json),
            RunAction::List { json } => list(&mut context, json),
            RunAction::Claim { run, json } => claim(&mut context, &run, json),
            RunAction::Migrate {
                run,
                all_runs,
                adopt_native,
            } => migrate(&mut context, run, all_runs, adopt_native),
            RunAction::Successor {
                run,
                source_incarnation,
                version,
                branch,
                base,
                source_baseline,
                baseline,
                worktree,
                confirm,
                json,
            } => successor(
                &mut context,
                &run,
                &source_incarnation,
                &version,
                &branch,
                &base,
                &source_baseline,
                &baseline,
                &worktree,
                confirm,
                json,
            ),
            RunAction::SuccessorAbort {
                run,
                source_incarnation,
                target_incarnation,
                confirm,
                json,
            } => successor_abort(
                &mut context,
                &run,
                &source_incarnation,
                &target_incarnation,
                confirm,
                json,
            ),
            RunAction::Transition { run, to } => transition(&mut context, &run, to),
            RunAction::Set {
                run,
                seed,
                plan,
                branch,
                base,
            } => set(&mut context, &run, seed, plan, branch, base),
            RunAction::Orientation { action } => orientation(&mut context, action),
            RunAction::Lane { action } => lane(&mut context, action),
            RunAction::Wave { action } => wave(&mut context, action),
            RunAction::Ledger { action } => ledger(&mut context, action),
            RunAction::Layout { run, repair, json } => layout(&mut context, &run, repair, json),
        }
    }
}

fn context(globals: CliGlobals) -> Result<ExecutionContext, CliError> {
    let cwd = std::env::current_dir()
        .map_err(|error| CliError::message(format!("cannot resolve current directory: {error}")))?;
    let mut inputs = ContextInputs::from_environment(cwd)
        .map_err(|error| CliError::message(error.to_string()))?;
    inputs.explicit_config = globals.config;
    inputs.verbosity = globals.verbosity;
    ExecutionContext::discover(inputs).map_err(|error| CliError::message(error.to_string()))
}

fn init(
    context: &mut ExecutionContext,
    requested: Option<String>,
    kind: RunKind,
    branch: &str,
    base: &str,
    version: &str,
    force: bool,
) -> Result<(), CliError> {
    let run = match requested {
        Some(run) => run,
        None => derive_id(if version.is_empty() { branch } else { version }, kind)?,
    };
    validate_id("run", &run)?;
    if !is_canonical(&run) && !force {
        return usage(format!(
            "non-canonical run id: {run:?} -- pass --force to override"
        ));
    }
    if !is_canonical(&run) && force {
        error_output(
            context,
            &format!("WARNING: {run:?} is a non-canonical run id, forced by --force."),
        )?;
    }
    let path = state_path(context, &run)?;
    if path.exists() {
        return missing_or_conflict(format!("run already exists: {run}"));
    }
    scaffold(context, &run)?;
    let authority = match crate::native_authority::register_run(context, &run, now_seconds(context))
    {
        Ok(authority) => authority,
        Err(error) => {
            let _ = fs::remove_dir_all(run_dir(context, &run)?);
            return Err(error);
        }
    };
    let mut state = RunState {
        schema_version: 1,
        run: run.clone(),
        run_incarnation: authority.run_incarnation.clone(),
        orientation_epoch: authority.orientation_epoch,
        kind: kind.into(),
        branch: branch.into(),
        base: base.into(),
        seed: String::new(),
        plan: String::new(),
        status: RunStatus::Planted.into(),
        lanes: Vec::new(),
        updated_at: now_seconds(context),
        extra: Default::default(),
    };
    state.updated_at = now_seconds(context);
    if let Err(source) = RunStore::new(&path).initialize(&state) {
        let _ = crate::native_authority::remove_run(context, &run);
        let _ = fs::remove_dir_all(run_dir(context, &run)?);
        return Err(store_error(source));
    }
    output(context, &path.display().to_string())
}

fn show(context: &mut ExecutionContext, run: &str, json: bool) -> Result<(), CliError> {
    let state = load(context, run)?;
    if json {
        return output(context, &state.to_canonical_json());
    }
    let mut lines = vec![
        format!("run: {}", state.run),
        format!("kind: {}", state.kind),
        format!("status: {}", state.status),
        format!("branch: {}", empty_dash(&state.branch)),
        format!("base: {}", empty_dash(&state.base)),
        format!("seed: {}", empty_dash(&state.seed)),
        format!("plan: {}", empty_dash(&state.plan)),
        format!("lanes: {}", state.lanes.len()),
    ];
    for lane in state.lanes {
        lines.push(format!(
            "  {}: {}{}",
            lane.id,
            lane.state,
            if lane.accepted_commit.is_some() && !lane.merged {
                " PENDING-MERGE"
            } else {
                ""
            }
        ));
    }
    output(context, &lines.join("\n"))
}

fn list(context: &mut ExecutionContext, json: bool) -> Result<(), CliError> {
    let mut runs = run_names(context, true)?;
    runs.sort();
    if json {
        return output(
            context,
            &serde_json::to_string(&runs).map_err(|error| CliError::message(error.to_string()))?,
        );
    }
    output(context, &runs.join("\n"))
}

fn claim(context: &mut ExecutionContext, run: &str, json: bool) -> Result<(), CliError> {
    let state = load(context, run)?;
    if state.schema_version > 1 {
        return usage(format!(
            "run {run} is schema_version {}, newer than this CLI supports claiming (max 1)",
            state.schema_version
        ));
    }
    let path = state_path(context, run)?;
    if json {
        return output(context, &serde_json::to_string_pretty(&serde_json::json!({"run": state.run, "schema_version": state.schema_version, "status": state.status, "lane_count": state.lanes.len(), "path": path})).map_err(|error| CliError::message(error.to_string()))?);
    }
    output(
        context,
        &format!(
            "claimed {} (schema {}, status {}, {} lane(s)): {}",
            state.run,
            state.schema_version,
            state.status,
            state.lanes.len(),
            path.display()
        ),
    )
}

fn orientation(context: &mut ExecutionContext, action: OrientationAction) -> Result<(), CliError> {
    let (value, json_output) = match action {
        OrientationAction::Pre { run, json } => {
            (crate::orientation::create_pre(context, &run, json)?, json)
        }
        OrientationAction::Post { run, json } => {
            (crate::orientation::create_post(context, &run, json)?, json)
        }
    };
    let rendered = if json_output {
        serde_json::to_string(&value).map_err(|error| CliError::message(error.to_string()))?
    } else {
        format!(
            "orientation {}: accepted={}",
            value["phase"].as_str().unwrap_or("unknown"),
            value["accepted"].as_bool().unwrap_or(false)
        )
    };
    output(context, &rendered)
}

/// Move a run to a new lifecycle status.
///
/// `planned` and `executing` are GUARDED: `orientation` re-verifies the run
/// incarnation, both checkpoints, the source ledger, the Critic verdict, and
/// the persisted post bytes while holding the run lock. A caller-authored
/// `status` value is not authority. The other three are ordinary lifecycle
/// marks and are written directly.
fn transition(
    context: &mut ExecutionContext,
    run: &str,
    to: shepherd::run::RunStatus,
) -> Result<(), CliError> {
    use shepherd::run::RunStatus;
    match to {
        RunStatus::Planned => crate::orientation::transition_planned(context, run)?,
        RunStatus::Executing => crate::orientation::transition_executing(context, run)?,
        other => update(context, run, |state| {
            state.status = other.into();
            Ok(())
        })?,
    }
    output(context, &format!("updated {run}"))
}

/// Set a run's identity fields.
///
/// This cannot express a status transition, by construction. It used to: `set`
/// took `--status` alongside the identity fields and routed two of the five
/// values to guarded transitions through an `if status ==` ladder, so one
/// subcommand was both a state machine and an unchecked field assignment. The
/// type system draws that line now instead of a conditional.
fn set(
    context: &mut ExecutionContext,
    run: &str,
    seed: Option<String>,
    plan: Option<String>,
    branch: Option<String>,
    base: Option<String>,
) -> Result<(), CliError> {
    if seed.is_none() && plan.is_none() && branch.is_none() && base.is_none() {
        return usage(
            "nothing to set (pass --seed, --plan, --branch, and/or --base; \
             use `run transition` to change status)",
        );
    }
    update(context, run, |state| {
        for (field, value) in [
            (&mut state.seed, seed),
            (&mut state.plan, plan),
            (&mut state.branch, branch),
            (&mut state.base, base),
        ] {
            if let Some(value) = value {
                *field = value;
            }
        }
        Ok(())
    })?;
    output(context, &format!("updated {run}"))
}

fn lane(context: &mut ExecutionContext, action: LaneAction) -> Result<(), CliError> {
    match action {
        LaneAction::Add {
            run,
            lane,
            plan,
            worktree,
            branch,
        } => {
            validate_id("lane", &lane)?;
            let run_dir = run_dir(context, &run)?;
            update(context, &run, |state| {
                if state.lanes.iter().any(|entry| entry.id == lane) {
                    return Err(RunStoreError::mutation(format!(
                        "lane already registered: {lane}"
                    )));
                }
                state.lanes.push(LaneState {
                    id: lane.clone(),
                    plan: if plan.is_empty() {
                        format!("lanes/{lane}/plan.md")
                    } else {
                        plan
                    },
                    worktree,
                    branch,
                    state: LaneStatus::Pending.into(),
                    accepted_commit: None,
                    merged: false,
                    updated_at: 0,
                    extra: Default::default(),
                });
                Ok(())
            })?;
            create_dir_safe(&run_dir.join("lanes").join(&lane))?;
            output(context, &format!("lane {lane} registered in {run}"))
        }
        LaneAction::Set { run, lane, state } => {
            update(context, &run, |document| {
                let Some(entry) = document.lanes.iter_mut().find(|entry| entry.id == lane) else {
                    return Err(RunStoreError::mutation(format!(
                        "no such lane: {lane} in run {run}"
                    )));
                };
                entry.state = state.into();
                Ok(())
            })?;
            output(context, &format!("lane {lane} -> {state}"))
        }
    }
}

fn wave(context: &mut ExecutionContext, action: WaveAction) -> Result<(), CliError> {
    match action {
        WaveAction::Accept { run, lane, commit } => {
            if commit.is_empty() {
                return usage("--commit must be non-empty");
            }
            update(context, &run, |state| {
                let Some(entry) = state.lanes.iter_mut().find(|entry| entry.id == lane) else {
                    return Err(RunStoreError::mutation(format!(
                        "no such lane: {lane} in run {run}"
                    )));
                };
                entry.accepted_commit = Some(commit.clone());
                entry.merged = false;
                Ok(())
            })?;
            output(context, &format!("accepted {lane} @ {commit}"))
        }
        WaveAction::Merged { run, lane } => {
            let commit = update(context, &run, |state| {
                let Some(entry) = state.lanes.iter_mut().find(|entry| entry.id == lane) else {
                    return Err(RunStoreError::mutation(format!(
                        "no such lane: {lane} in run {run}"
                    )));
                };
                let Some(commit) = entry.accepted_commit.clone() else {
                    return Err(RunStoreError::mutation(format!(
                        "lane {lane} has no accepted commit to mark merged"
                    )));
                };
                entry.merged = true;
                Ok(commit)
            })?;
            output(context, &format!("merged {lane} @ {commit}"))
        }
        WaveAction::Pending { run, json } => pending(context, &run, json),
        WaveAction::Verify { run, wave, json } => verify(context, &run, wave, json),
    }
}

fn pending(context: &mut ExecutionContext, run: &str, json: bool) -> Result<(), CliError> {
    let state = load(context, run)?;
    let pending: Vec<_> = state
        .lanes
        .iter()
        .filter(|lane| lane.accepted_commit.is_some() && !lane.merged)
        .collect();
    let plan = fs::read_to_string(run_dir(context, run)?.join("plan.md")).unwrap_or_default();
    let declared = declared_lanes(&plan);
    let missing: Vec<_> = declared
        .into_iter()
        .filter(|lane| {
            !state
                .lanes
                .iter()
                .any(|entry| entry.id.eq_ignore_ascii_case(lane))
        })
        .collect();
    if json {
        output(context, &serde_json::to_string(&serde_json::json!({"pending": pending.iter().map(|lane| serde_json::json!({"lane": lane.id, "commit": lane.accepted_commit})).collect::<Vec<_>>(), "missing_lanes": missing, "ok": pending.is_empty() && missing.is_empty()})).map_err(|error| CliError::message(error.to_string()))?)?;
    } else {
        let rows = pending
            .iter()
            .map(|lane| {
                format!(
                    "{}\t{}",
                    lane.id,
                    lane.accepted_commit.as_deref().unwrap_or_default()
                )
            })
            .chain(
                missing
                    .iter()
                    .map(|lane| format!("{lane}\tMISSING-DECLARED-LANE")),
            )
            .collect::<Vec<_>>()
            .join("\n");
        if !rows.is_empty() {
            output(context, &rows)?;
        }
    }
    if pending.is_empty() && missing.is_empty() {
        Ok(())
    } else {
        Err(CliError::message_with_code(
            "wave pending: accepted work remains or declared lanes are missing",
            6,
        ))
    }
}

fn layout(
    context: &mut ExecutionContext,
    run: &str,
    repair: bool,
    json: bool,
) -> Result<(), CliError> {
    let path = state_path(context, run)?;
    if !path.is_file() {
        return no_such_run(context, run);
    }
    let base = run_dir(context, run)?;
    let mut missing = RUN_SUBDIRS
        .iter()
        .filter(|name| !base.join(name).is_dir())
        .copied()
        .collect::<Vec<_>>();
    let mut created = Vec::new();
    if repair {
        for name in &missing {
            create_dir_safe(&base.join(name))?;
            created.push(*name);
        }
        missing.retain(|name| !base.join(name).is_dir());
    }
    if json {
        output(context, &serde_json::to_string_pretty(&serde_json::json!({"run":run,"run_dir":base,"subdirs":RUN_SUBDIRS,"missing":missing,"created":created,"tracked_files_present":TRACKED_FILES.iter().filter(|name| base.join(name).is_file()).collect::<Vec<_>>(),"ok":missing.is_empty()})).map_err(|error| CliError::message(error.to_string()))?)?;
    } else {
        let mut lines = RUN_SUBDIRS
            .iter()
            .map(|name| {
                format!(
                    "{:<12}{}",
                    format!("{name}/"),
                    if missing.contains(name) {
                        "missing"
                    } else if created.contains(name) {
                        "created"
                    } else {
                        "ok"
                    }
                )
            })
            .collect::<Vec<_>>();
        lines.push(format!(
            "tracked artifacts: {}",
            TRACKED_FILES
                .iter()
                .filter(|name| base.join(name).is_file())
                .copied()
                .collect::<Vec<_>>()
                .join(", ")
        ));
        output(context, &lines.join("\n"))?;
    }
    if missing.is_empty() {
        Ok(())
    } else {
        Err(CliError::message_with_code(
            format!(
                "layout incomplete ({}) — re-run with --repair",
                missing.join(", ")
            ),
            6,
        ))
    }
}

fn rename(
    context: &mut ExecutionContext,
    old: &str,
    new: &str,
    announce: bool,
) -> Result<(), CliError> {
    validate_id("run", old)?;
    validate_id("run", new)?;
    crate::native_authority::ensure_no_pending_successor(context, old)?;
    if old == new {
        return usage("old and new run ids are identical");
    }
    let old_dir = run_dir(context, old)?;
    let new_dir = run_dir(context, new)?;
    if !old_dir.is_dir() {
        return Err(CliError::message_with_code(
            format!(
                "no such run directory: {old} (expected {})",
                old_dir.display()
            ),
            5,
        ));
    }
    if new_dir.exists() {
        return Err(CliError::message_with_code(
            format!("destination already exists: {new}"),
            5,
        ));
    }
    reject_symlink_path(&old_dir)?;
    reject_symlink_path(&new_dir)?;
    let registered = old_dir.join("run.json").is_file();
    fs::rename(&old_dir, &new_dir)
        .map_err(|error| CliError::message(format!("rename {old} -> {new}: {error}")))?;
    let native_renamed = if registered {
        match crate::native_authority::rename_run(context, old, new) {
            Ok(renamed) => renamed,
            Err(error) => {
                let _ = fs::rename(&new_dir, &old_dir);
                return Err(error);
            }
        }
    } else {
        false
    };
    if registered {
        let path = state_path(context, new)?;
        if let Err(source) = RunStore::new(&path).rewrite_from_raw(|bytes| {
            let mut state: RunState = serde_json::from_slice(bytes).map_err(|error| {
                RunStoreError::mutation(format!("run.json for {old} could not be read: {error}"))
            })?;
            if state.run != old {
                return Err(RunStoreError::mutation(format!(
                    "document run `{}` does not match renamed directory `{old}`",
                    state.run
                )));
            }
            state.run = new.into();
            let old_prefix = format!("runs/{old}/");
            let new_prefix = format!("runs/{new}/");
            if state.seed.starts_with(&old_prefix) {
                state.seed = format!("{new_prefix}{}", &state.seed[old_prefix.len()..]);
            }
            if state.plan.starts_with(&old_prefix) {
                state.plan = format!("{new_prefix}{}", &state.plan[old_prefix.len()..]);
            }
            state.updated_at = now_seconds(context);
            Ok((state, ()))
        }) {
            if native_renamed {
                let _ = crate::native_authority::rename_run(context, new, old);
            }
            let _ = fs::rename(&new_dir, &old_dir);
            return Err(store_error(source));
        }
    }
    if announce {
        output(
            context,
            &format!("renamed {old} -> {new}: {}", new_dir.display()),
        )
    } else {
        Ok(())
    }
}

fn canonicalize(
    context: &mut ExecutionContext,
    run: Option<String>,
    all_runs: bool,
    dry_run: bool,
) -> Result<(), CliError> {
    if context.config.branching.sprint_slug_pattern != "v{X}{Y}{Z}-dev{N}"
        || context.config.branching.patch_slug_pattern != "v{X}{Y}{Z}"
    {
        return usage(
            "run canonicalize supports only the default slug patterns; configured patterns require a native pattern parser before this route can be promoted",
        );
    }
    if run.is_some() == all_runs {
        return usage("pass exactly one of <run> or --all");
    }
    let targets = match run {
        Some(value) => vec![value],
        None => run_names(context, false)?,
    };
    if targets.is_empty() {
        return output(context, "no runs to canonicalize");
    }
    let mut lines = Vec::new();
    for target in targets {
        let source = run_dir(context, &target)?;
        if !source.is_dir() {
            return Err(CliError::message_with_code(
                format!(
                    "no such run directory: {target} (expected {})",
                    source.display()
                ),
                5,
            ));
        }
        if is_canonical(&target) {
            lines.push(format!("{target}: already canonical"));
            continue;
        }
        let Some(candidate) = canonical_suggestion(&target) else {
            lines.push(format!("{target}: no recognizable canonical form -- fix manually with: shepherd run rename {target} <new-id>"));
            continue;
        };
        if run_dir(context, &candidate)?.exists() {
            lines.push(format!("{target}: canonical form {candidate:?} already exists -- refusing to overwrite, fix manually"));
            continue;
        }
        if dry_run {
            lines.push(format!(
                "{target} -> {candidate} (dry run, no changes made)"
            ));
            continue;
        }
        rename(context, &target, &candidate, false)?;
        lines.push(format!(
            "{target} -> {candidate}: {}",
            run_dir(context, &candidate)?.display()
        ));
    }
    if !lines.is_empty() {
        output(context, &lines.join("\n"))
    } else {
        Ok(())
    }
}

fn migrate(
    context: &mut ExecutionContext,
    run: Option<String>,
    all_runs: bool,
    adopt_native: bool,
) -> Result<(), CliError> {
    if run.is_some() == all_runs {
        return usage("pass exactly one of <run> or --all");
    }
    if adopt_native && all_runs {
        return usage(
            "--adopt-native requires one explicit run; bulk authority adoption is refused",
        );
    }
    let targets = match run {
        Some(value) => vec![value],
        None => run_names(context, true)?,
    };
    if targets.is_empty() {
        return output(context, "no runs to migrate");
    }
    let mut lines = Vec::new();
    for run in targets {
        let path = state_path(context, &run)?;
        if adopt_native {
            let incarnation = RunStore::new(&path)
                .update_with_access(|state, access| {
                    crate::native_authority::ensure_no_pending_successor(context, &run).map_err(
                        |source| {
                            RunStoreError::mutation(
                                source
                                    .message_text()
                                    .unwrap_or("incomplete successor transition"),
                            )
                        },
                    )?;
                    crate::native_authority::adopt_legacy_state(
                        context,
                        &run,
                        state,
                        access,
                        now_seconds(context),
                    )
                    .map_err(|source| {
                        RunStoreError::mutation(
                            source
                                .message_text()
                                .unwrap_or("native legacy adoption failed"),
                        )
                    })
                })
                .map_err(store_error)?;
            lines.push(format!(
                "adopted {run} into native incarnation {incarnation}: {}",
                path.display()
            ));
            continue;
        }
        let applied = RunStore::new(&path)
            .rewrite_from_raw(|bytes| {
                crate::native_authority::ensure_no_pending_successor(context, &run).map_err(
                    |source| {
                        RunStoreError::mutation(
                            source
                                .message_text()
                                .unwrap_or("incomplete successor transition"),
                        )
                    },
                )?;
                let raw: serde_json::Value = serde_json::from_slice(bytes).map_err(|error| {
                    RunStoreError::mutation(format!(
                        "run.json for {run} could not be read: {error}"
                    ))
                })?;
                let (mut document, applied) = normalize_document(raw).map_err(|error| {
                    RunStoreError::mutation(error.message_text().unwrap_or("run migration failed"))
                })?;
                if document.run != run {
                    return Err(RunStoreError::mutation(format!(
                        "run.json for {run} has mismatched run identity `{}`",
                        document.run
                    )));
                }
                document.updated_at = now_seconds(context);
                Ok((document, applied))
            })
            .map_err(store_error)?;
        let migration_note = if applied.is_empty() {
            "no changes".to_owned()
        } else {
            applied.join(", ")
        };
        lines.push(format!(
            "migrated {run} ({migration_note}): {}",
            path.display()
        ));
    }
    output(context, &lines.join("\n"))
}

#[allow(clippy::too_many_arguments)]
fn successor(
    context: &mut ExecutionContext,
    run: &str,
    source_incarnation: &str,
    version: &str,
    branch: &str,
    base: &str,
    source_baseline: &str,
    baseline: &str,
    worktree: &Path,
    confirm: bool,
    json: bool,
) -> Result<(), CliError> {
    if !confirm {
        return usage(
            "run successor is an archival mutation; pass --confirm after verifying every identity field",
        );
    }
    validate_id("run", run)?;
    if !is_lower_hex(source_incarnation, 32) {
        return usage("source incarnation must be the exact 32-character lowercase Native id");
    }
    if !exact_commit(source_baseline) {
        return usage("source baseline must be an exact lowercase 40-hex commit");
    }
    if !exact_commit(baseline) {
        return usage("baseline must be an exact lowercase 40-hex commit");
    }
    let parsed = semver::Version::parse(version)
        .map_err(|error| usage_error(format!("invalid semantic version `{version}`: {error}")))?;
    if parsed.to_string() != version {
        return usage(format!(
            "version must use canonical SemVer spelling: expected `{parsed}`, observed `{version}`"
        ));
    }
    let id_kind = if parsed.pre.is_empty() {
        RunKind::PatchArc
    } else {
        RunKind::Sprint
    };
    let expected_run = derive_id(&format!("v{parsed}"), id_kind)?;
    if expected_run != run {
        return usage(format!(
            "version mismatch: version `{version}` derives run `{expected_run}`, observed `{run}`"
        ));
    }
    let expected_branch = format!("v{parsed}");
    if branch != expected_branch {
        return usage(format!(
            "version and branch disagree: expected branch `{expected_branch}`, observed `{branch}`"
        ));
    }
    let (worktree, base_commit) = validate_successor_worktree(
        context,
        worktree,
        run,
        branch,
        base,
        source_baseline,
        baseline,
    )?;
    let worktree_identity = worktree.display().to_string();
    let native = crate::native_authority::successor_state(context, run, source_incarnation)?;
    let target_incarnation = match &native {
        crate::native_authority::SuccessorNativeState::Current(_) => {
            crate::native_authority::new_incarnation()
        }
        crate::native_authority::SuccessorNativeState::Pending(pending) => {
            pending.target_incarnation.clone()
        }
        crate::native_authority::SuccessorNativeState::Complete(retired) => {
            retired.successor_incarnation.clone()
        }
    };
    let state_path = state_path(context, run)?;
    let run_dir = run_dir(context, run)?;
    let source_archive = run_dir
        .join(".incarnations")
        .join(source_incarnation)
        .join("source");
    let now = now_seconds(context);
    let report = RunStore::new(&state_path)
        .with_exclusive_optional(SUCCESSOR_DOCUMENT_LIMIT, |active, access| {
            successor_locked(
                context,
                &native,
                active,
                run,
                source_incarnation,
                &target_incarnation,
                version,
                branch,
                base,
                &base_commit,
                source_baseline,
                baseline,
                &worktree_identity,
                &run_dir,
                &source_archive,
                access,
                now,
            )
            .map_err(|error| {
                RunStoreError::mutation(
                    error
                        .message_text()
                        .unwrap_or("same-version successor transition failed"),
                )
            })
        })
        .map_err(store_error)?;
    if json {
        output(
            context,
            &serde_json::to_string(&report)
                .map_err(|error| CliError::message(error.to_string()))?,
        )
    } else {
        output(
            context,
            &format!(
                "successor {}: {} -> {}; source archived at {}",
                run,
                source_incarnation,
                target_incarnation,
                source_archive.display()
            ),
        )
    }
}

fn successor_abort(
    context: &mut ExecutionContext,
    run: &str,
    source_incarnation: &str,
    target_incarnation: &str,
    confirm: bool,
    json: bool,
) -> Result<(), CliError> {
    if !confirm {
        return usage("run successor-abort preserves failed evidence and requires --confirm");
    }
    validate_id("run", run)?;
    for (label, value) in [
        ("source incarnation", source_incarnation),
        ("target incarnation", target_incarnation),
    ] {
        if !is_lower_hex(value, 32) {
            return usage(format!(
                "{label} must be an exact 32-character lowercase Native id"
            ));
        }
    }
    let pending = match crate::native_authority::successor_state(context, run, source_incarnation)?
    {
        crate::native_authority::SuccessorNativeState::Pending(pending) => pending,
        crate::native_authority::SuccessorNativeState::Current(_) => {
            return usage("run has no pending successor intent to abort");
        }
        crate::native_authority::SuccessorNativeState::Complete(_) => {
            return usage("completed successors are immutable and cannot be aborted");
        }
    };
    if pending.target_incarnation != target_incarnation {
        return usage(format!(
            "target incarnation mismatch: expected `{}`, observed `{target_incarnation}`",
            pending.target_incarnation
        ));
    }
    let run_dir = run_dir(context, run)?;
    let state_path = state_path(context, run)?;
    let incarnation_root = run_dir.join(".incarnations").join(source_incarnation);
    let expected_archive = incarnation_root.join("source");
    if pending.source_archive != expected_archive.display().to_string() {
        return Err(CliError::message(
            "pending successor archive does not match the canonical incarnation path",
        ));
    }
    let aborted = RunStore::new(state_path)
        .with_exclusive_optional(SUCCESSOR_DOCUMENT_LIMIT, |active, access| {
            abort_successor_locked(
                context,
                &pending,
                active,
                access,
                &run_dir,
                &expected_archive,
                &incarnation_root,
            )
            .map_err(|error| {
                RunStoreError::mutation(
                    error
                        .message_text()
                        .unwrap_or("same-version successor abort failed"),
                )
            })
        })
        .map_err(store_error)?;
    let report = serde_json::json!({
        "schema": "shepherd.run-successor-abort/1",
        "run": run,
        "source_incarnation": source_incarnation,
        "target_incarnation": target_incarnation,
        "failure_archive": aborted.failure_archive,
        "aborted_at": aborted.aborted_at,
    });
    if json {
        output(
            context,
            &serde_json::to_string(&report)
                .map_err(|error| CliError::message(error.to_string()))?,
        )
    } else {
        output(
            context,
            &format!(
                "aborted unactivated successor {source_incarnation} -> {target_incarnation}; failed evidence preserved at {}",
                incarnation_root.display()
            ),
        )
    }
}

fn abort_successor_locked(
    context: &ExecutionContext,
    pending: &crate::native_authority::PendingSuccessor,
    active: Option<&RunState>,
    access: &crate::run_store::RunAccess<'_>,
    run_dir: &Path,
    source_archive: &Path,
    incarnation_root: &Path,
) -> Result<crate::native_authority::AbortedSuccessor, CliError> {
    let active_source =
        active.is_some_and(|state| state.run_incarnation == pending.source_incarnation);
    let source_complete = if active_source {
        let observed = successor_tree_digest(run_dir, true)?;
        observed == pending.source_archive_sha256
    } else {
        false
    };
    if !source_complete {
        if successor_tree_digest(source_archive, false)? != pending.source_archive_sha256 {
            return Err(CliError::message(
                "archived source does not match the pending successor intent",
            ));
        }
        let staging = prepare_abort_staging(pending, source_archive, incarnation_root)?;
        if !active_source {
            evacuate_unactivated_target(pending, active, access, run_dir, incarnation_root)?;
        }
        publish_staged_source(pending, source_archive, &staging, access, run_dir)?;
    }
    let restored = successor_tree_digest(run_dir, true)?;
    if restored != pending.source_archive_sha256 {
        return Err(CliError::message(format!(
            "restored source differs during successor abort: expected {}, observed {restored}",
            pending.source_archive_sha256
        )));
    }
    crate::native_authority::abort_successor(
        context,
        pending,
        incarnation_root,
        now_seconds(context),
    )
}

fn prepare_abort_staging(
    pending: &crate::native_authority::PendingSuccessor,
    source_archive: &Path,
    incarnation_root: &Path,
) -> Result<PathBuf, CliError> {
    let staging_root = incarnation_root.join(format!("abort-{}", pending.abort_nonce));
    create_dir_safe(&staging_root)?;
    for attempt in 0..16_u8 {
        let candidate = staging_root.join(format!("restore-{attempt}"));
        if successor_entry_exists(&candidate, "successor abort staging")? {
            if fs::symlink_metadata(&candidate)
                .is_ok_and(|metadata| metadata.is_dir() && !metadata.file_type().is_symlink())
                && successor_tree_digest(&candidate, false)
                    .is_ok_and(|digest| digest == pending.source_archive_sha256)
            {
                return Ok(candidate);
            }
            continue;
        }
        create_dir_safe(&candidate)?;
        copy_successor_tree(source_archive, &candidate)?;
        let observed = successor_tree_digest(&candidate, false)?;
        if observed != pending.source_archive_sha256 {
            return Err(CliError::message(format!(
                "successor abort staging differs from source archive: expected {}, observed {observed}",
                pending.source_archive_sha256
            )));
        }
        return Ok(candidate);
    }
    Err(CliError::message(format!(
        "successor abort exhausted 16 authenticated staging attempts under {}",
        staging_root.display()
    )))
}

fn publish_staged_source(
    pending: &crate::native_authority::PendingSuccessor,
    source_archive: &Path,
    staging: &Path,
    access: &crate::run_store::RunAccess<'_>,
    run_dir: &Path,
) -> Result<(), CliError> {
    let staging_relative = staging
        .strip_prefix(run_dir)
        .map_err(|_| CliError::message("successor abort staging escaped the held run"))?;
    let mut entries = Vec::new();
    for entry in fs::read_dir(source_archive)
        .map_err(|error| CliError::message(format!("read source archive: {error}")))?
    {
        if entries.len() >= SUCCESSOR_ENTRY_LIMIT {
            return usage("source archive entry count changed beyond the successor bound");
        }
        entries.push(
            entry.map_err(|error| {
                CliError::message(format!("read source archive entry: {error}"))
            })?,
        );
    }
    entries.sort_by_key(std::fs::DirEntry::file_name);
    for entry in entries {
        let name = entry.file_name();
        let archived = entry.path();
        let expected = successor_content_witness(&archived)?;
        let destination = run_dir.join(&name);
        if successor_entry_exists(&destination, "restored source destination")? {
            if successor_content_witness(&destination)? != expected {
                return usage(format!(
                    "foreign restored-source destination is preserved: {}",
                    destination.display()
                ));
            }
            continue;
        }
        let staged = staging.join(&name);
        if successor_content_witness(&staged)? != expected {
            return Err(CliError::message(format!(
                "authenticated successor abort staging changed: {}",
                staged.display()
            )));
        }
        access
            .restore_entry(staging_relative, &name)
            .map_err(store_error)?;
        if successor_content_witness(&destination)? != expected {
            return Err(CliError::message(format!(
                "restored source entry changed during atomic publication: {}",
                destination.display()
            )));
        }
    }
    if successor_tree_digest(run_dir, true)? != pending.source_archive_sha256 {
        return Err(CliError::message(
            "published source tree does not match pending successor custody",
        ));
    }
    Ok(())
}

fn successor_content_witness(path: &Path) -> Result<String, CliError> {
    let metadata = fs::symlink_metadata(path).map_err(|error| {
        CliError::message(format!(
            "inspect successor content {}: {error}",
            path.display()
        ))
    })?;
    if metadata.file_type().is_symlink() {
        return usage("successor content witness refuses symlinks");
    }
    let digest = if metadata.is_dir() {
        successor_tree_digest(path, false)?
    } else if metadata.is_file() {
        sha256_hex(
            &crate::dispatch_service::read_path_nofollow(
                path,
                usize::try_from(SUCCESSOR_FILE_LIMIT).expect("successor file limit fits usize"),
            )
            .map_err(|error| CliError::message(error.to_string()))?,
        )
    } else {
        return usage("successor content witness requires a regular file or directory");
    };
    Ok(format!(
        "{}:{}:{}:{digest}",
        if metadata.is_dir() {
            "directory"
        } else {
            "file"
        },
        metadata.len(),
        successor_file_mode(&metadata)
    ))
}

fn evacuate_unactivated_target(
    pending: &crate::native_authority::PendingSuccessor,
    active: Option<&RunState>,
    access: &crate::run_store::RunAccess<'_>,
    run_dir: &Path,
    incarnation_root: &Path,
) -> Result<(), CliError> {
    if let Some(state) = active
        && state.run_incarnation != pending.target_incarnation
    {
        return usage(format!(
            "successor abort found an unrelated active incarnation: expected source `{}` or target `{}`, observed `{}`",
            pending.source_incarnation, pending.target_incarnation, state.run_incarnation
        ));
    }
    let failure_target = incarnation_root
        .join("failed-targets")
        .join(&pending.target_incarnation);
    let locations = ["dispatch", "lanes", "run.json"]
        .map(|name| {
            Ok((
                name,
                successor_entry_exists(&run_dir.join(name), "active target entry")?,
                successor_entry_exists(&failure_target.join(name), "failed target entry")?,
            ))
        })
        .into_iter()
        .collect::<Result<Vec<_>, CliError>>()?;
    if locations
        .iter()
        .all(|(_, active, failed)| !active && !failed)
    {
        return Ok(());
    }
    if locations.iter().any(|(_, active, failed)| active == failed) {
        return usage("successor abort found ambiguous partial target custody");
    }
    for (name, active, _) in &locations {
        let path = if *active {
            run_dir.join(name)
        } else {
            failure_target.join(name)
        };
        if *name == "run.json" {
            let bytes = read_successor_file(
                &path,
                "unactivated successor state",
                SUCCESSOR_DOCUMENT_LIMIT,
            )?;
            if sha256_hex(&bytes) != pending.target_state_sha256 {
                return usage("successor abort refuses a modified target run state");
            }
        } else if !directory_is_empty(&path)? {
            return usage("successor abort refuses target lanes or dispatch authority");
        }
    }
    create_dir_safe(&failure_target)?;
    let failure_relative = failure_target
        .strip_prefix(run_dir)
        .map_err(|_| CliError::message("failed-target archive escaped the held run directory"))?;
    for (name, active, _) in locations {
        if active {
            access
                .archive_entry(std::ffi::OsStr::new(name), failure_relative)
                .map_err(store_error)?;
        }
    }
    for entry in fs::read_dir(run_dir)
        .map_err(|error| CliError::message(format!("inspect evacuated target: {error}")))?
    {
        let name = entry
            .map_err(|error| CliError::message(format!("inspect evacuated target: {error}")))?
            .file_name();
        if name != "run.lock" && name != ".incarnations" {
            return usage(format!(
                "successor abort refuses unowned target entry `{}`",
                name.to_string_lossy()
            ));
        }
    }
    Ok(())
}

fn directory_is_empty(path: &Path) -> Result<bool, CliError> {
    let mut entries = fs::read_dir(path)
        .map_err(|error| CliError::message(format!("inspect {}: {error}", path.display())))?;
    Ok(entries
        .next()
        .transpose()
        .map_err(|error| CliError::message(format!("inspect {}: {error}", path.display())))?
        .is_none())
}

fn copy_successor_tree(source: &Path, destination: &Path) -> Result<(), CliError> {
    let mut budget = SuccessorTreeBudget::default();
    copy_successor_tree_bounded(source, source, destination, 0, &mut budget)
}

fn copy_successor_tree_bounded(
    source_root: &Path,
    source: &Path,
    destination: &Path,
    depth: usize,
    budget: &mut SuccessorTreeBudget,
) -> Result<(), CliError> {
    if depth > SUCCESSOR_DEPTH_LIMIT {
        return usage("source archive depth changed beyond the successor bound");
    }
    let mut entries = Vec::new();
    for entry in fs::read_dir(source)
        .map_err(|error| CliError::message(format!("read source archive: {error}")))?
    {
        budget.entries = budget.entries.saturating_add(1);
        if budget.entries > SUCCESSOR_ENTRY_LIMIT {
            return usage("source archive entry count changed beyond the successor bound");
        }
        entries.push(
            entry.map_err(|error| {
                CliError::message(format!("read source archive entry: {error}"))
            })?,
        );
    }
    entries.sort_by_key(std::fs::DirEntry::file_name);
    for entry in entries {
        let source_path = entry.path();
        let relative = source_path
            .strip_prefix(source_root)
            .map_err(|_| CliError::message("source archive path escaped its root"))?
            .to_str()
            .ok_or_else(|| CliError::message("source archive path is not portable UTF-8"))?;
        if relative.len() > SUCCESSOR_PATH_LIMIT {
            return usage("source archive path changed beyond the successor bound");
        }
        let destination_path = destination.join(entry.file_name());
        let destination_exists =
            successor_entry_exists(&destination_path, "restored source destination")?;
        let metadata = fs::symlink_metadata(&source_path).map_err(|error| {
            CliError::message(format!("inspect archived source entry: {error}"))
        })?;
        if metadata.file_type().is_symlink() {
            return usage("archived source contains a symlink");
        }
        if metadata.is_dir() {
            if destination_exists {
                let destination_metadata =
                    fs::symlink_metadata(&destination_path).map_err(|error| {
                        CliError::message(format!("inspect restored directory: {error}"))
                    })?;
                if destination_metadata.file_type().is_symlink() || !destination_metadata.is_dir() {
                    return usage("partial successor abort restored a conflicting directory");
                }
            } else {
                create_dir_safe(&destination_path)?;
            }
            copy_successor_tree_bounded(
                source_root,
                &source_path,
                &destination_path,
                depth + 1,
                budget,
            )?;
            continue;
        }
        if !metadata.is_file() || metadata.len() > SUCCESSOR_FILE_LIMIT {
            return usage("archived source contains an unsafe or oversized entry");
        }
        budget.bytes = budget.bytes.saturating_add(metadata.len());
        if budget.bytes > SUCCESSOR_TOTAL_LIMIT {
            return usage("source archive bytes changed beyond the successor bound");
        }
        let bytes = crate::dispatch_service::read_path_nofollow(
            &source_path,
            usize::try_from(SUCCESSOR_FILE_LIMIT).expect("successor file limit fits usize"),
        )
        .map_err(|error| CliError::message(error.to_string()))?;
        if destination_exists {
            let existing = crate::dispatch_service::read_path_nofollow(
                &destination_path,
                usize::try_from(SUCCESSOR_FILE_LIMIT).expect("successor file limit fits usize"),
            )
            .map_err(|error| CliError::message(error.to_string()))?;
            let existing_metadata = fs::symlink_metadata(&destination_path).map_err(|error| {
                CliError::message(format!("inspect partially restored source: {error}"))
            })?;
            if existing != bytes
                || successor_file_mode(&existing_metadata) != successor_file_mode(&metadata)
            {
                return usage("partial successor abort restored conflicting source bytes");
            }
            continue;
        }
        let mut options = fs::OpenOptions::new();
        options.write(true).create_new(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            options.mode(successor_file_mode(&metadata));
        }
        let mut output = options
            .open(&destination_path)
            .map_err(|error| CliError::message(format!("create restored source entry: {error}")))?;
        output
            .write_all(&bytes)
            .and_then(|()| output.sync_all())
            .map_err(|error| CliError::message(format!("write restored source entry: {error}")))?;
    }
    Ok(())
}

#[allow(clippy::too_many_arguments)]
fn successor_locked(
    context: &ExecutionContext,
    observed_native: &crate::native_authority::SuccessorNativeState,
    active: Option<&RunState>,
    run: &str,
    source_incarnation: &str,
    target_incarnation: &str,
    version: &str,
    branch: &str,
    base: &str,
    base_commit: &str,
    source_baseline: &str,
    baseline: &str,
    worktree_identity: &str,
    run_dir: &Path,
    source_archive: &Path,
    access: &crate::run_store::RunAccess<'_>,
    now: i64,
) -> Result<serde_json::Value, CliError> {
    let archive_state_path = source_archive.join("run.json");
    let active_source_path = run_dir.join("run.json");
    let source_state_path = if successor_entry_exists(&archive_state_path, "archived source state")?
    {
        &archive_state_path
    } else {
        &active_source_path
    };
    let source_bytes = read_successor_file(
        source_state_path,
        "source run state",
        SUCCESSOR_DOCUMENT_LIMIT,
    )?;
    let source_state: RunState = serde_json::from_slice(&source_bytes)
        .map_err(|error| usage_error(format!("source run state is invalid: {error}")))?;
    validate_successor_source(&source_state, run, source_incarnation, branch, base)?;
    let active_probes_path = source_state_path
        .parent()
        .ok_or_else(|| usage_error("source run state has no parent"))?
        .join("plan-probes.json");
    let archived_probes_path = source_archive.join("plan-probes.json");
    let probes_path = if successor_entry_exists(&active_probes_path, "active source plan probes")? {
        &active_probes_path
    } else {
        &archived_probes_path
    };
    let probes: serde_json::Value = serde_json::from_slice(&read_successor_file(
        probes_path,
        "source plan probes",
        SUCCESSOR_DOCUMENT_LIMIT,
    )?)
    .map_err(|error| usage_error(format!("source plan probes are invalid: {error}")))?;
    let observed_source_baseline = probes
        .get("baseline")
        .and_then(serde_json::Value::as_str)
        .ok_or_else(|| usage_error("source plan probes have no exact baseline"))?;
    if observed_source_baseline != source_baseline {
        return usage(format!(
            "source baseline mismatch: expected `{source_baseline}`, observed `{observed_source_baseline}`"
        ));
    }
    let source_state_sha256 = sha256_hex(&source_bytes);
    let source_archive_sha256 = match observed_native {
        crate::native_authority::SuccessorNativeState::Current(_) => {
            successor_tree_digest(run_dir, true)?
        }
        crate::native_authority::SuccessorNativeState::Pending(pending) => {
            pending.source_archive_sha256.clone()
        }
        crate::native_authority::SuccessorNativeState::Complete(retired) => {
            let observed = successor_tree_digest(source_archive, false)?;
            if observed != retired.source_archive_sha256 {
                return Err(CliError::message(format!(
                    "archived source evidence changed: expected {}, observed {observed}",
                    retired.source_archive_sha256
                )));
            }
            observed
        }
    };

    if let crate::native_authority::SuccessorNativeState::Complete(retired) = observed_native {
        let active = active
            .ok_or_else(|| CliError::message("completed successor has no active run state"))?;
        if active.run_incarnation != retired.successor_incarnation
            || source_state_sha256 != retired.source_state_sha256
        {
            return Err(CliError::message(
                "completed successor identity or archived source does not match Native history",
            ));
        }
        let spec = crate::native_authority::SuccessorSpec {
            run: run.into(),
            source_incarnation: source_incarnation.into(),
            target_incarnation: target_incarnation.into(),
            version: version.into(),
            branch: branch.into(),
            base: base.into(),
            base_commit: base_commit.into(),
            source_baseline: source_baseline.into(),
            baseline: baseline.into(),
            worktree_identity: worktree_identity.into(),
            source_state_sha256,
            source_archive_sha256,
            target_state_sha256: retired.target_state_sha256.clone(),
            source_archive: source_archive.display().to_string(),
        };
        if !crate::native_authority::successor_matches_spec(retired, &spec) {
            return usage("completed successor identity differs from the exact replay request");
        }
        return Ok(successor_report(run, retired));
    }

    let created_at = match observed_native {
        crate::native_authority::SuccessorNativeState::Current(authority) => {
            if active.is_none_or(|state| {
                state.run_incarnation != authority.run_incarnation
                    || state.orientation_epoch != authority.orientation_epoch
            }) {
                return Err(CliError::message(
                    "active run state does not match Native source authority",
                ));
            }
            now
        }
        crate::native_authority::SuccessorNativeState::Pending(pending) => pending.created_at,
        crate::native_authority::SuccessorNativeState::Complete(_) => unreachable!(),
    };
    let target_state = RunState {
        schema_version: 1,
        run: run.into(),
        run_incarnation: target_incarnation.into(),
        orientation_epoch: 0,
        kind: source_state.kind.clone(),
        branch: branch.into(),
        base: base.into(),
        seed: String::new(),
        plan: String::new(),
        status: RunStatus::Planted.into(),
        lanes: Vec::new(),
        updated_at: created_at,
        extra: Default::default(),
    };
    let target_bytes = canonical_state_bytes(&target_state);
    let spec = crate::native_authority::SuccessorSpec {
        run: run.into(),
        source_incarnation: source_incarnation.into(),
        target_incarnation: target_incarnation.into(),
        version: version.into(),
        branch: branch.into(),
        base: base.into(),
        base_commit: base_commit.into(),
        source_baseline: source_baseline.into(),
        baseline: baseline.into(),
        worktree_identity: worktree_identity.into(),
        source_state_sha256,
        source_archive_sha256,
        target_state_sha256: sha256_hex(&target_bytes),
        source_archive: source_archive.display().to_string(),
    };
    let pending = match observed_native {
        crate::native_authority::SuccessorNativeState::Current(authority) => {
            crate::native_authority::prepare_successor(context, authority, &spec, created_at)?
        }
        crate::native_authority::SuccessorNativeState::Pending(pending) => {
            if pending.base_commit != base_commit {
                return usage(format!(
                    "source base commit changed during successor recovery: expected `{}`, observed `{base_commit}`",
                    pending.base_commit
                ));
            }
            if !crate::native_authority::pending_matches_spec(pending, &spec) {
                return usage("pending successor identity differs from the exact replay request");
            }
            pending.clone()
        }
        crate::native_authority::SuccessorNativeState::Complete(_) => unreachable!(),
    };
    rotate_successor_run(
        run_dir,
        source_archive,
        source_incarnation,
        &pending.target_incarnation,
        &pending.source_state_sha256,
        &pending.source_archive_sha256,
        &pending.target_state_sha256,
        &target_state,
        access,
    )?;
    let retired = crate::native_authority::complete_successor(context, &pending, now)?;
    Ok(successor_report(run, &retired))
}

fn validate_successor_source(
    state: &RunState,
    run: &str,
    source_incarnation: &str,
    branch: &str,
    base: &str,
) -> Result<(), CliError> {
    if state.run != run {
        return usage(format!(
            "source run mismatch: expected `{run}`, observed `{}`",
            state.run
        ));
    }
    if state.run_incarnation != source_incarnation {
        return usage(format!(
            "source incarnation mismatch: expected `{source_incarnation}`, observed `{}`",
            state.run_incarnation
        ));
    }
    if !state.status.is(RunStatus::Planted) {
        return usage(format!(
            "source run must be planted for same-version recovery, observed `{}`",
            state.status
        ));
    }
    if state.branch != branch {
        return usage(format!(
            "source branch mismatch: expected `{branch}`, observed `{}`",
            state.branch
        ));
    }
    if state.base != base {
        return usage(format!(
            "source base mismatch: expected `{base}`, observed `{}`",
            state.base
        ));
    }
    Ok(())
}

fn validate_successor_worktree(
    context: &ExecutionContext,
    requested: &Path,
    run: &str,
    branch: &str,
    base: &str,
    source_baseline: &str,
    baseline: &str,
) -> Result<(PathBuf, String), CliError> {
    if !requested.is_absolute() {
        return usage("worktree must be an explicit absolute canonical path");
    }
    let worktree = fs::canonicalize(requested)
        .map_err(|error| usage_error(format!("cannot canonicalize selected worktree: {error}")))?;
    if worktree != requested {
        return usage(format!(
            "worktree path is not canonical: expected `{}`, observed `{}`",
            worktree.display(),
            requested.display()
        ));
    }
    reject_symlink_path(&worktree).map_err(|error| {
        usage_error(format!(
            "selected worktree path is unsafe: {}",
            error.message_text().unwrap_or("symlinked path")
        ))
    })?;
    let observed_root = PathBuf::from(git_value(
        &worktree,
        &["rev-parse", "--show-toplevel"],
        "selected worktree root",
    )?);
    let observed_root = fs::canonicalize(&observed_root).map_err(|error| {
        usage_error(format!(
            "cannot canonicalize observed worktree root: {error}"
        ))
    })?;
    if observed_root != worktree {
        return usage(format!(
            "selected worktree mismatch: expected `{}`, observed `{}`",
            worktree.display(),
            observed_root.display()
        ));
    }
    let primary_common = canonical_git_common_dir(&context.primary_root)?;
    let selected_common = canonical_git_common_dir(&worktree)?;
    if primary_common != selected_common {
        return usage(format!(
            "selected worktree belongs to another project: expected git common dir `{}`, observed `{}`",
            primary_common.display(),
            selected_common.display()
        ));
    }
    let observed_branch = git_value(
        &worktree,
        &["symbolic-ref", "--quiet", "--short", "HEAD"],
        "selected worktree branch (detached HEAD is unsupported)",
    )?;
    if observed_branch != branch {
        return usage(format!(
            "selected worktree branch mismatch: expected `{branch}`, observed `{observed_branch}`"
        ));
    }
    let observed_head = git_value(
        &worktree,
        &["rev-parse", "--verify", "HEAD"],
        "selected worktree HEAD",
    )?;
    if observed_head != baseline {
        return usage(format!(
            "baseline mismatch: expected `{baseline}`, observed `{observed_head}`"
        ));
    }
    let observed_base = git_value(
        &worktree,
        &["rev-parse", "--verify", &format!("{base}^{{commit}}")],
        "source base ref",
    )?;
    if !exact_commit(&observed_base) {
        return usage(format!(
            "base `{base}` did not resolve to an exact commit: observed `{observed_base}`"
        ));
    }
    let source_kind = git_value(
        &worktree,
        &["cat-file", "-t", source_baseline],
        "source baseline",
    )?;
    if source_kind != "commit" {
        return usage(format!(
            "source baseline is not a commit: observed `{source_kind}`"
        ));
    }
    let ancestor = Command::new("git")
        .args(["merge-base", "--is-ancestor", source_baseline, baseline])
        .current_dir(&worktree)
        .status()
        .map_err(|error| CliError::message(format!("inspect source baseline ancestry: {error}")))?;
    if !ancestor.success() {
        return usage("source baseline is not an ancestor of the requested successor baseline");
    }
    ensure_successor_product_clean(&worktree, run)?;
    Ok((worktree, observed_base))
}

fn ensure_successor_product_clean(worktree: &Path, run: &str) -> Result<(), CliError> {
    let run_exclusion = format!(":(exclude).shepherd/runs/{run}/**");
    for cached in [false, true] {
        let mut command = Command::new("git");
        command.arg("diff").arg("--quiet");
        if cached {
            command.arg("--cached");
        }
        let status = command
            .arg("--")
            .arg(".")
            .arg(&run_exclusion)
            .arg(":(exclude).shepherd/native-orientation-registry.json")
            .current_dir(worktree)
            .status()
            .map_err(|error| {
                CliError::message(format!("inspect uncommitted product state: {error}"))
            })?;
        match status.code() {
            Some(0) => {}
            Some(1) => {
                return usage(
                    "uncommitted product changes exist outside the selected run evidence",
                );
            }
            _ => {
                return Err(CliError::message(
                    "git could not inspect uncommitted product state",
                ));
            }
        }
    }
    let untracked = Command::new("git")
        .args(["ls-files", "--others", "--exclude-standard", "-z"])
        .current_dir(worktree)
        .output()
        .map_err(|error| CliError::message(format!("inspect untracked product state: {error}")))?;
    if !untracked.status.success() {
        return Err(CliError::message(
            "git could not inspect untracked product state",
        ));
    }
    let run_prefix = format!(".shepherd/runs/{run}/");
    if untracked
        .stdout
        .split(|byte| *byte == 0)
        .filter(|path| !path.is_empty())
        .any(|path| std::str::from_utf8(path).map_or(true, |path| !path.starts_with(&run_prefix)))
    {
        return usage("uncommitted product files exist outside the selected run evidence");
    }
    Ok(())
}

fn canonical_git_common_dir(worktree: &Path) -> Result<PathBuf, CliError> {
    let path = PathBuf::from(git_value(
        worktree,
        &["rev-parse", "--path-format=absolute", "--git-common-dir"],
        "git common directory",
    )?);
    fs::canonicalize(&path)
        .map_err(|error| usage_error(format!("cannot canonicalize git common directory: {error}")))
}

fn git_value(root: &Path, args: &[&str], label: &str) -> Result<String, CliError> {
    let result = Command::new("git")
        .args(args)
        .current_dir(root)
        .output()
        .map_err(|error| CliError::message(format!("inspect {label}: {error}")))?;
    if !result.status.success() {
        let detail = String::from_utf8_lossy(&result.stderr).trim().to_owned();
        return usage(format!(
            "cannot resolve {label}: {}",
            if detail.is_empty() {
                "git refused the query"
            } else {
                &detail
            }
        ));
    }
    Ok(String::from_utf8_lossy(&result.stdout).trim().to_owned())
}

#[allow(clippy::too_many_arguments)]
fn rotate_successor_run(
    run_dir: &Path,
    source_archive: &Path,
    source_incarnation: &str,
    target_incarnation: &str,
    source_state_sha256: &str,
    source_archive_sha256: &str,
    target_state_sha256: &str,
    target_state: &RunState,
    access: &crate::run_store::RunAccess<'_>,
) -> Result<(), CliError> {
    create_dir_safe(source_archive)?;
    let archive_relative = source_archive
        .strip_prefix(run_dir)
        .map_err(|_| CliError::message("successor archive is outside the held run directory"))?;
    let archived_state = source_archive.join("run.json");
    if !successor_entry_exists(&archived_state, "archived source state")? {
        let active_state = run_dir.join("run.json");
        let active: RunState = serde_json::from_slice(&read_successor_file(
            &active_state,
            "active source state",
            SUCCESSOR_DOCUMENT_LIMIT,
        )?)
        .map_err(|error| usage_error(format!("active source state is invalid: {error}")))?;
        if active.run_incarnation != source_incarnation {
            return usage(format!(
                "active source incarnation mismatch: expected `{source_incarnation}`, observed `{}`",
                active.run_incarnation
            ));
        }
        let mut entries = Vec::new();
        for entry in fs::read_dir(run_dir)
            .map_err(|error| CliError::message(format!("read source run directory: {error}")))?
        {
            if entries.len() >= SUCCESSOR_ENTRY_LIMIT {
                return usage("source run entry count changed beyond the successor bound");
            }
            entries.push(
                entry.map_err(|error| {
                    CliError::message(format!("read source run entry: {error}"))
                })?,
            );
        }
        entries.sort_by_key(std::fs::DirEntry::file_name);
        let entries = entries
            .into_iter()
            .filter(|entry| {
                let name = entry.file_name();
                name != "run.lock" && name != "run.json" && name != ".incarnations"
            })
            .collect::<Vec<_>>();
        if successor_tree_digest(run_dir, true)? != source_archive_sha256 {
            return Err(CliError::message(
                "source run evidence changed after successor intent publication",
            ));
        }
        for entry in &entries {
            let name = entry.file_name();
            if successor_entry_exists(&source_archive.join(&name), "successor archive destination")?
            {
                return Err(CliError::message(format!(
                    "successor archive destination already exists: {}",
                    source_archive.join(name).display()
                )));
            }
        }
        if successor_entry_exists(&archived_state, "successor archive destination")? {
            return Err(CliError::message(format!(
                "successor archive destination already exists: {}",
                archived_state.display()
            )));
        }
        let mut witnesses = Vec::with_capacity(entries.len());
        for entry in &entries {
            witnesses.push((entry.file_name(), successor_entry_witness(&entry.path())?));
        }
        let state_witness = successor_entry_witness(&active_state)?;
        for (name, expected) in witnesses {
            let source = run_dir.join(&name);
            let observed = successor_entry_witness(&source)?;
            if observed != expected {
                return Err(CliError::message(format!(
                    "source entry changed before archival move: {}",
                    source.display()
                )));
            }
            access
                .archive_entry(&name, archive_relative)
                .map_err(store_error)?;
            if successor_entry_witness(&source_archive.join(&name))? != expected {
                return Err(CliError::message(format!(
                    "archived entry identity changed during move: {}",
                    source_archive.join(name).display()
                )));
            }
        }
        if successor_entry_witness(&active_state)? != state_witness {
            return Err(CliError::message(
                "source run state changed before archival move",
            ));
        }
        access
            .archive_entry(std::ffi::OsStr::new("run.json"), archive_relative)
            .map_err(store_error)?;
        if successor_entry_witness(&archived_state)? != state_witness {
            return Err(CliError::message(
                "archived run state identity changed during move",
            ));
        }
    }
    let archived_bytes = read_successor_file(
        &archived_state,
        "archived source state",
        SUCCESSOR_DOCUMENT_LIMIT,
    )?;
    let archived: RunState = serde_json::from_slice(&archived_bytes)
        .map_err(|error| usage_error(format!("archived source state is invalid: {error}")))?;
    if archived.run_incarnation != source_incarnation
        || sha256_hex(&archived_bytes) != source_state_sha256
    {
        return Err(CliError::message(
            "archived source state does not match the prepared successor intent",
        ));
    }
    let observed_archive_sha256 = successor_tree_digest(source_archive, false)?;
    if observed_archive_sha256 != source_archive_sha256 {
        return Err(CliError::message(format!(
            "archived source evidence differs from the prepared intent: expected {source_archive_sha256}, observed {observed_archive_sha256}"
        )));
    }
    let active_path = run_dir.join("run.json");
    if successor_entry_exists(&active_path, "active successor state")? {
        let active_bytes = read_successor_file(
            &active_path,
            "active successor state",
            SUCCESSOR_DOCUMENT_LIMIT,
        )?;
        let active: RunState = serde_json::from_slice(&active_bytes)
            .map_err(|error| usage_error(format!("active successor state is invalid: {error}")))?;
        if active.run_incarnation != target_incarnation
            || sha256_hex(&active_bytes) != target_state_sha256
        {
            return Err(CliError::message(
                "active successor state does not match the prepared Native intent",
            ));
        }
    } else {
        create_dir_safe(&run_dir.join("lanes"))?;
        create_dir_safe(&run_dir.join("dispatch"))?;
        target_state
            .store(&active_path)
            .map_err(|error| CliError::message(error.to_string()))?;
        let written = read_successor_file(
            &active_path,
            "written successor state",
            SUCCESSOR_DOCUMENT_LIMIT,
        )?;
        if sha256_hex(&written) != target_state_sha256 {
            return Err(CliError::message(
                "written successor state differs from the prepared Native intent",
            ));
        }
    }
    Ok(())
}

fn successor_entry_witness(path: &Path) -> Result<String, CliError> {
    let metadata = fs::symlink_metadata(path).map_err(|error| {
        CliError::message(format!(
            "inspect successor source entry {}: {error}",
            path.display()
        ))
    })?;
    if metadata.file_type().is_symlink() {
        return usage(format!(
            "successor source entry is a symlink: {}",
            path.display()
        ));
    }
    let kind = if metadata.is_dir() {
        "directory"
    } else if metadata.is_file() {
        "file"
    } else {
        return usage(format!(
            "successor source entry is not regular: {}",
            path.display()
        ));
    };
    let content = if metadata.is_dir() {
        successor_tree_digest(path, false)?
    } else {
        sha256_hex(
            &crate::dispatch_service::read_path_nofollow(
                path,
                usize::try_from(SUCCESSOR_FILE_LIMIT).expect("successor file limit fits usize"),
            )
            .map_err(|error| CliError::message(error.to_string()))?,
        )
    };
    #[cfg(unix)]
    let filesystem = {
        use std::os::unix::fs::MetadataExt;
        format!("{}:{}", metadata.dev(), metadata.ino())
    };
    #[cfg(windows)]
    let filesystem = {
        let (volume, file) = crate::safe_fs::windows_path_id(path)
            .map_err(|error| CliError::message(error.to_string()))?;
        format!("{volume}:{file}")
    };
    #[cfg(all(not(unix), not(windows)))]
    let filesystem = format!(
        "{}",
        metadata
            .modified()
            .ok()
            .and_then(|value| value.duration_since(std::time::UNIX_EPOCH).ok())
            .map_or(0, |value| value.as_nanos())
    );
    Ok(format!(
        "{kind}:{filesystem}:{}:{}:{content}",
        metadata.len(),
        successor_file_mode(&metadata)
    ))
}

fn successor_entry_exists(path: &Path, label: &str) -> Result<bool, CliError> {
    match fs::symlink_metadata(path) {
        Ok(metadata) if metadata.file_type().is_symlink() => usage(format!(
            "{label} is a symlink and is refused: {}",
            path.display()
        )),
        Ok(_) => Ok(true),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
        Err(error) => Err(CliError::message(format!(
            "inspect {label} {}: {error}",
            path.display()
        ))),
    }
}

fn read_successor_file(path: &Path, label: &str, limit: usize) -> Result<Vec<u8>, CliError> {
    crate::dispatch_service::read_path_nofollow(path, limit).map_err(|error| {
        CliError::message(format!(
            "cannot read bounded no-follow {label} {}: {error}",
            path.display()
        ))
    })
}

fn successor_tree_digest(root: &Path, active_run: bool) -> Result<String, CliError> {
    let mut digest = Sha256::new();
    let mut budget = SuccessorTreeBudget::default();
    digest_successor_tree(root, root, active_run, 0, &mut budget, &mut digest)?;
    Ok(digest
        .finalize()
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect())
}

#[derive(Default)]
struct SuccessorTreeBudget {
    entries: usize,
    bytes: u64,
}

fn digest_successor_tree(
    root: &Path,
    directory: &Path,
    active_run: bool,
    depth: usize,
    budget: &mut SuccessorTreeBudget,
    digest: &mut Sha256,
) -> Result<(), CliError> {
    if depth > SUCCESSOR_DEPTH_LIMIT {
        return usage(format!(
            "successor evidence exceeds the maximum depth of {SUCCESSOR_DEPTH_LIMIT}"
        ));
    }
    let reader = fs::read_dir(directory).map_err(|error| {
        CliError::message(format!(
            "read successor evidence directory {}: {error}",
            directory.display()
        ))
    })?;
    let mut entries = Vec::new();
    for entry in reader {
        budget.entries = budget
            .entries
            .checked_add(1)
            .ok_or_else(|| usage_error("successor evidence entry count overflowed"))?;
        if budget.entries > SUCCESSOR_ENTRY_LIMIT {
            return usage(format!(
                "successor evidence exceeds the maximum entry count of {SUCCESSOR_ENTRY_LIMIT}"
            ));
        }
        entries.push(entry.map_err(|error| {
            CliError::message(format!("read successor evidence entry: {error}"))
        })?);
    }
    entries.sort_by_key(std::fs::DirEntry::file_name);
    for entry in entries {
        let name = entry.file_name();
        if active_run && directory == root && (name == "run.lock" || name == ".incarnations") {
            continue;
        }
        let path = entry.path();
        let relative = path
            .strip_prefix(root)
            .map_err(|_| CliError::message("successor evidence escaped its archive root"))?;
        let relative = relative
            .to_str()
            .ok_or_else(|| CliError::message("successor evidence path is not portable UTF-8"))?;
        if relative.len() > SUCCESSOR_PATH_LIMIT {
            return usage(format!(
                "successor evidence path exceeds the maximum {SUCCESSOR_PATH_LIMIT} bytes: {relative}"
            ));
        }
        let before = fs::symlink_metadata(&path).map_err(|error| {
            CliError::message(format!(
                "inspect successor evidence {}: {error}",
                path.display()
            ))
        })?;
        if before.file_type().is_symlink() {
            return usage(format!(
                "successor evidence contains a symlink: {}",
                path.display()
            ));
        }
        if before.is_dir() {
            digest.update(b"directory\0");
            digest.update(relative.as_bytes());
            digest.update(b"\0");
            digest_successor_tree(root, &path, false, depth + 1, budget, digest)?;
            continue;
        }
        if !before.is_file() {
            return usage(format!(
                "successor evidence contains a non-regular entry: {}",
                path.display()
            ));
        }
        if before.len() > SUCCESSOR_FILE_LIMIT {
            return usage(format!(
                "successor evidence file exceeds the maximum {SUCCESSOR_FILE_LIMIT} bytes: {relative}"
            ));
        }
        budget.bytes = budget
            .bytes
            .checked_add(before.len())
            .ok_or_else(|| usage_error("successor evidence byte count overflowed"))?;
        if budget.bytes > SUCCESSOR_TOTAL_LIMIT {
            return usage(format!(
                "successor evidence exceeds the maximum total of {SUCCESSOR_TOTAL_LIMIT} bytes"
            ));
        }
        let bytes = crate::dispatch_service::read_path_nofollow(
            &path,
            usize::try_from(SUCCESSOR_FILE_LIMIT).expect("successor file limit fits usize"),
        )
        .map_err(|error| {
            CliError::message(format!(
                "cannot read bounded no-follow successor evidence {}: {error}",
                path.display()
            ))
        })?;
        digest.update(b"file\0");
        digest.update(relative.as_bytes());
        digest.update(b"\0");
        digest.update(before.len().to_le_bytes());
        digest.update(successor_file_mode(&before).to_le_bytes());
        digest.update(&bytes);
        let after = fs::symlink_metadata(&path).map_err(|error| {
            CliError::message(format!(
                "reinspect successor evidence {}: {error}",
                path.display()
            ))
        })?;
        if before.len() != u64::try_from(bytes.len()).unwrap_or(u64::MAX)
            || before.len() != after.len()
            || before.modified().ok() != after.modified().ok()
            || successor_file_mode(&before) != successor_file_mode(&after)
        {
            return Err(CliError::message(format!(
                "successor evidence changed while it was read: {}",
                path.display()
            )));
        }
        digest.update(b"\0");
    }
    Ok(())
}

fn successor_file_mode(metadata: &fs::Metadata) -> u32 {
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        metadata.permissions().mode() & 0o777
    }
    #[cfg(not(unix))]
    {
        u32::from(metadata.permissions().readonly())
    }
}

fn canonical_state_bytes(state: &RunState) -> Vec<u8> {
    let mut bytes = state.to_canonical_json().into_bytes();
    bytes.push(b'\n');
    bytes
}

fn is_lower_hex(value: &str, length: usize) -> bool {
    value.len() == length
        && value
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}

fn exact_commit(value: &str) -> bool {
    is_lower_hex(value, 40)
}

fn successor_report(
    run: &str,
    retired: &crate::native_authority::RetiredRunIncarnation,
) -> serde_json::Value {
    serde_json::json!({
        "schema": "shepherd.run-successor/1",
        "run": run,
        "source_incarnation": retired.run_incarnation,
        "target_incarnation": retired.successor_incarnation,
        "version": retired.version,
        "branch": retired.branch,
        "base": retired.base,
        "base_commit": retired.base_commit,
        "source_baseline": retired.source_baseline,
        "baseline": retired.baseline,
        "worktree_identity": retired.worktree_identity,
        "source_state_sha256": retired.source_state_sha256,
        "source_archive_sha256": retired.source_archive_sha256,
        "target_state_sha256": retired.target_state_sha256,
        "source_archive": retired.source_archive,
        "succeeded_at": retired.succeeded_at,
    })
}

fn ledger(context: &mut ExecutionContext, action: LedgerAction) -> Result<(), CliError> {
    match action {
        LedgerAction::Path { run, check } => {
            let run = resolve_active(context, run)?;
            let path = run_dir(context, &run)?.join(LEDGER_FILE);
            output(context, &crate::interface::canonical_display(&path))?;
            if check && local_ledger_exists(context, &run)? {
                return Err(CliError::message_with_code(
                    format!(
                        "divergent local ledger copy for {run}; use {}",
                        path.display()
                    ),
                    3,
                ));
            }
            Ok(())
        }
        LedgerAction::Check { run, json } => {
            ledger_check(context, resolve_active(context, run)?, json)
        }
    }
}

fn ledger_check(context: &mut ExecutionContext, run: String, json: bool) -> Result<(), CliError> {
    let primary = run_dir(context, &run)?.join(LEDGER_FILE);
    let primary_text = fs::read_to_string(&primary).map_err(|_| {
        CliError::message_with_code(
            format!("no ledger for run {run} (expected {})", primary.display()),
            5,
        )
    })?;
    let mut divergences = Vec::new();
    let git_output = Command::new("git")
        .args(["worktree", "list", "--porcelain"])
        .current_dir(&context.primary_root)
        .output();
    match git_output {
        Ok(output) if output.status.success() => {
            let worktree_text = String::from_utf8_lossy(&output.stdout);
            let worktrees = worktree_text
                .lines()
                .filter_map(|line| line.strip_prefix("worktree "))
                .map(str::to_owned)
                .collect::<Vec<_>>();
            for worktree in worktrees.into_iter().skip(1) {
                let candidate = Path::new(&worktree)
                    .join(context.namespace.file_name().unwrap_or_default())
                    .join("runs")
                    .join(&run)
                    .join(LEDGER_FILE);
                if let Ok(text) = fs::read_to_string(candidate) {
                    for row in normalized_rows(&text) {
                        if !normalized_rows(&primary_text).contains(&row) {
                            divergences.push(serde_json::json!({"worktree":worktree,"row":row}));
                        }
                    }
                }
            }
        }
        _ => {}
    }
    if json {
        output(context, &serde_json::to_string(&serde_json::json!({"run":run,"divergences":divergences,"ok":divergences.is_empty()})).map_err(|error| CliError::message(error.to_string()))?)?;
    } else if !divergences.is_empty() {
        output(
            context,
            &divergences
                .iter()
                .map(|value| {
                    format!(
                        "{}\t{}",
                        value["worktree"].as_str().unwrap_or_default(),
                        value["row"].as_str().unwrap_or_default()
                    )
                })
                .collect::<Vec<_>>()
                .join("\n"),
        )?;
    }
    if divergences.is_empty() {
        Ok(())
    } else {
        Err(CliError::message_with_code("worktree ledger divergence", 7))
    }
}

fn verify(
    context: &mut ExecutionContext,
    run: &str,
    wave: Option<u32>,
    json: bool,
) -> Result<(), CliError> {
    let base = run_dir(context, run)?;
    if !base.join("lanes").is_dir() {
        return Err(CliError::message_with_code(
            format!(
                "no lane plans for run {run} (expected {})",
                base.join("lanes").display()
            ),
            5,
        ));
    }
    let mut steps = Vec::new();
    for lane in fs::read_dir(base.join("lanes"))
        .map_err(|_| CliError::message_with_code(format!("no lane plans for run {run}"), 5))?
    {
        let lane = lane.map_err(|error| CliError::message(error.to_string()))?;
        let text = fs::read_to_string(lane.path().join("plan.md")).unwrap_or_default();
        steps.extend(parse_plan_steps(&text));
    }
    if let Some(wave) = wave {
        steps.retain(|step| step.wave == wave);
    }
    let ledger = fs::read_to_string(base.join(LEDGER_FILE)).unwrap_or_default();
    let rows = parse_ledger(&ledger);
    let malformed = ledger
        .lines()
        .enumerate()
        .filter(|(_, line)| {
            !line.trim().is_empty()
                && !line.trim().starts_with('#')
                && parse_ledger_line(line).is_none()
        })
        .map(|(line_no, line)| format!("MALFORMED-ROW\tline {}: {}", line_no + 1, line))
        .collect::<Vec<_>>();
    let mut findings = malformed;
    let mut rendered = Vec::new();
    for step in &steps {
        let winner = rows.iter().rev().find(|row| {
            row.lane == step.lane
                && row.wave == step.wave
                && (row.step.is_none() || row.step == Some(step.step))
        });
        rendered.push(format!(
            "{}\t{}\t{}",
            step.id(),
            winner.map_or("-", |row| row.verdict.as_str()),
            winner.map_or("-", |row| row.raw.as_str())
        ));
        match winner {
            None => findings.push(format!("NO-VERDICT\t{} has no ledger verdict", step.id())),
            Some(row) if row.verdict != "PASS" => findings.push(format!(
                "UNRESOLVED-VERDICT\t{} resolves to {}",
                step.id(),
                row.verdict
            )),
            _ => {}
        }
    }
    for row in &rows {
        if !steps.iter().any(|step| {
            step.lane == row.lane
                && step.wave == row.wave
                && (row.step.is_none() || row.step == Some(step.step))
        }) {
            findings.push(format!("ORPHAN-VERDICT\t{}", row.raw));
        }
    }
    if json {
        output(context, &serde_json::to_string(&serde_json::json!({"run":run,"wave":wave,"steps":rendered,"findings":findings,"ok":findings.is_empty()})).map_err(|error| CliError::message(error.to_string()))?)?;
    } else {
        if !rendered.is_empty() {
            output(context, &rendered.join("\n"))?;
        }
        if !findings.is_empty() {
            output(context, &format!("\nFINDINGS:\n{}", findings.join("\n")))?;
        }
    }
    if findings.is_empty() {
        Ok(())
    } else {
        Err(CliError::message_with_code("wave verification findings", 6))
    }
}

fn load(context: &ExecutionContext, run: &str) -> Result<RunState, CliError> {
    let path = state_path(context, run)?;
    reject_symlink_path(&path)?;
    RunStore::new(&path).load().map_err(|error| match error {
        RunStoreError::Io { source, .. } if source.kind() == std::io::ErrorKind::NotFound => {
            no_such_run_error(context, run)
        }
        other => store_error(other),
    })
}

fn update<T>(
    context: &ExecutionContext,
    run: &str,
    mutation: impl FnOnce(&mut RunState) -> Result<T, RunStoreError>,
) -> Result<T, CliError> {
    let path = state_path(context, run)?;
    reject_symlink_path(&path)?;
    RunStore::new(&path)
        .update(|state| {
            crate::native_authority::ensure_no_pending_successor(context, run).map_err(
                |source| {
                    RunStoreError::mutation(
                        source
                            .message_text()
                            .unwrap_or("incomplete successor transition"),
                    )
                },
            )?;
            let value = mutation(state)?;
            state.updated_at = now_seconds(context);
            Ok(value)
        })
        .map_err(|error| match error {
            RunStoreError::Io { source, .. } if source.kind() == std::io::ErrorKind::NotFound => {
                no_such_run_error(context, run)
            }
            RunStoreError::Mutation(message) if message.starts_with("no such lane:") => {
                CliError::message_with_code(message, 5)
            }
            RunStoreError::Mutation(message) => CliError::message_with_code(message, 2),
            other => store_error(other),
        })
}

fn state_path(context: &ExecutionContext, run: &str) -> Result<PathBuf, CliError> {
    validate_id("run", run)?;
    Ok(context.runs_root.join(run).join("run.json"))
}

fn run_dir(context: &ExecutionContext, run: &str) -> Result<PathBuf, CliError> {
    validate_id("run", run)?;
    Ok(context.runs_root.join(run))
}

fn scaffold(context: &ExecutionContext, run: &str) -> Result<(), CliError> {
    let base = run_dir(context, run)?;
    create_dir_safe(&base)?;
    for name in RUN_SUBDIRS {
        create_dir_safe(&base.join(name))?;
    }
    Ok(())
}

fn create_dir_safe(path: &Path) -> Result<(), CliError> {
    if let Some(parent) = path.parent() {
        reject_symlink_path(parent)?;
    }
    fs::create_dir_all(path).map_err(|error| {
        CliError::message(format!("create directory {}: {error}", path.display()))
    })?;
    reject_symlink_path(path)
}

fn reject_symlink_path(path: &Path) -> Result<(), CliError> {
    use std::path::Component;

    let mut cursor = PathBuf::new();
    for component in path.components() {
        cursor.push(component.as_os_str());
        // A prefix or root is not a stat-able entry and cannot be a symlink.
        // On Windows, `symlink_metadata` on a bare `\\?\C:` returns
        // `Incorrect function. (os error 1)`, so probing it turned every run
        // command into `ERROR: inspect \\?\C:: Incorrect function.`
        if matches!(component, Component::Prefix(_) | Component::RootDir) {
            continue;
        }
        match fs::symlink_metadata(&cursor) {
            Ok(metadata) if metadata.file_type().is_symlink() => {
                return Err(CliError::message(format!(
                    "refusing symlinked run path: {}",
                    cursor.display()
                )));
            }
            Ok(_) => {}
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => break,
            Err(error) => {
                return Err(CliError::message(format!(
                    "inspect {}: {error}",
                    cursor.display()
                )));
            }
        }
    }
    Ok(())
}

fn run_names(context: &ExecutionContext, registered_only: bool) -> Result<Vec<String>, CliError> {
    match fs::read_dir(&context.runs_root) {
        Ok(entries) => entries
            .map(|entry| entry.map_err(|error| CliError::message(error.to_string())))
            .filter_map(|entry| match entry {
                Ok(entry) if entry.file_type().map(|kind| kind.is_dir()).unwrap_or(false) => {
                    entry.file_name().into_string().ok().map(Ok)
                }
                Ok(_) => None,
                Err(error) => Some(Err(error)),
            })
            .filter(|name| {
                name.as_ref().is_ok_and(|name| {
                    !registered_only || context.runs_root.join(name).join("run.json").is_file()
                })
            })
            .collect(),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
        Err(error) => Err(CliError::message(format!(
            "read runs directory {}: {error}",
            context.runs_root.display()
        ))),
    }
}

fn resolve_active(context: &ExecutionContext, run: Option<String>) -> Result<String, CliError> {
    if let Some(run) = run {
        return Ok(run);
    }
    let mut candidates = Vec::new();
    for run in run_names(context, true)? {
        let path = state_path(context, &run)?;
        let modified = fs::metadata(&path)
            .and_then(|metadata| metadata.modified())
            .ok();
        if load(context, &run).is_ok_and(|state| state.status.is(RunStatus::Executing)) {
            candidates.push((modified, run));
        }
    }
    candidates.sort_by_key(|candidate| std::cmp::Reverse(candidate.0));
    candidates.into_iter().next().map(|(_, run)| run).ok_or_else(|| CliError::message_with_code("no <run> given and no active run found (a runs/*/run.json with status: \"executing\") -- pass <run> explicitly", 2))
}

fn local_ledger_exists(context: &ExecutionContext, run: &str) -> Result<bool, CliError> {
    let output = Command::new("git")
        .args(["rev-parse", "--show-toplevel"])
        .current_dir(std::env::current_dir().map_err(|error| CliError::message(error.to_string()))?)
        .output()
        .map_err(|error| CliError::message(format!("git worktree lookup: {error}")))?;
    if !output.status.success() {
        return Ok(false);
    }
    let current = PathBuf::from(String::from_utf8_lossy(&output.stdout).trim());
    if current == context.primary_root {
        return Ok(false);
    }
    let namespace = context
        .namespace
        .file_name()
        .ok_or_else(|| CliError::message("configured namespace has no basename"))?;
    Ok(current
        .join(namespace)
        .join("runs")
        .join(run)
        .join(LEDGER_FILE)
        .is_file())
}

fn normalize_document(raw: serde_json::Value) -> Result<(RunState, Vec<String>), CliError> {
    let mut object = raw
        .as_object()
        .cloned()
        .ok_or_else(|| usage_error("run.json root must be an object"))?;
    let mut applied = Vec::new();
    if let (true, Some(run)) = (!object.contains_key("run"), object.remove("run_id")) {
        object.insert("run".into(), run);
        applied.push("run_id -> run".into());
    }
    if let Some(lanes) = object
        .get("lanes")
        .cloned()
        .and_then(|lanes| lanes.as_object().cloned())
    {
        let mut entries = lanes
            .iter()
            .map(|(id, value)| {
                let mut value = value.as_object().cloned().unwrap_or_default();
                // Case-fold the dict key on the way in. Historical documents
                // used upper-case lane ids (`V01-STATIC-CONTRACT`), and
                // `RunStore::validate_id` -- a WRITE-side naming rule --
                // rejects them on read. That single mismatch is what stopped
                // `run migrate` from repairing the very documents it exists to
                // repair: the migrator produced a correct list and the store
                // then refused it. Normalizing here is the repair.
                let normalized = id.to_ascii_lowercase();
                value.insert("id".into(), serde_json::Value::String(normalized));
                serde_json::Value::Object(value)
            })
            .collect::<Vec<_>>();
        entries.sort_by(|left, right| left["id"].as_str().cmp(&right["id"].as_str()));
        object.insert("lanes".into(), serde_json::Value::Array(entries));
        applied.push("lanes dict -> list".into());
        if lanes
            .keys()
            .any(|id| id.chars().any(|c| c.is_ascii_uppercase()))
        {
            applied.push("lane ids case-folded to lower-case".into());
        }
    }
    match object.get("updated_at").cloned() {
        Some(value) if !value.is_i64() && !value.is_u64() => {
            object.insert(
                "updated_at".into(),
                serde_json::Value::from(coerce_epoch(&value)),
            );
            applied.push("updated_at -> epoch".into());
        }
        _ => {}
    }
    serde_json::from_value(serde_json::Value::Object(object))
        .map(|state| (state, applied))
        .map_err(|error| usage_error(format!("run.json schema validation failed: {error}")))
}

fn coerce_epoch(value: &serde_json::Value) -> i64 {
    value
        .as_i64()
        .or_else(|| value.as_u64().and_then(|value| i64::try_from(value).ok()))
        .or_else(|| {
            value
                .as_f64()
                .and_then(|value| value.trunc().to_string().parse::<i64>().ok())
        })
        .or_else(|| value.as_str().and_then(|value| value.parse::<i64>().ok()))
        .unwrap_or(0)
}

fn declared_lanes(plan: &str) -> Vec<String> {
    let mut lines = plan.lines();
    for line in lines.by_ref() {
        if line
            .trim_start_matches('#')
            .trim()
            .eq_ignore_ascii_case("lane projection")
        {
            break;
        }
    }
    let mut result = Vec::new();
    let mut header = false;
    let mut separator = false;
    for line in lines {
        let trimmed = line.trim();
        if trimmed.starts_with('#') {
            break;
        }
        if !trimmed.starts_with('|') {
            if header {
                break;
            }
            continue;
        }
        let cells = trimmed
            .trim_matches('|')
            .split('|')
            .map(str::trim)
            .collect::<Vec<_>>();
        if !header {
            header = true;
            if cells.first().is_none_or(|cell| {
                !cell
                    .trim_matches(['`', '*', ' '])
                    .eq_ignore_ascii_case("lane_id")
            }) {
                break;
            }
            continue;
        }
        if !separator {
            separator = true;
            if cells
                .iter()
                .all(|cell| cell.trim_matches([':', '-']).is_empty())
            {
                continue;
            }
        }
        if let Some(id) = cells
            .first()
            .map(|cell| cell.trim_matches(['`', '*', ' ']).to_ascii_lowercase())
            .filter(|id| !id.is_empty())
        {
            result.push(id);
        }
    }
    result
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct Step {
    wave: u32,
    lane: u32,
    step: u32,
}
impl Step {
    fn id(&self) -> String {
        format!("W{}-L{}-S{}", self.wave, self.lane, self.step)
    }
}
#[derive(Clone, Debug)]
struct LedgerRow {
    lane: u32,
    wave: u32,
    step: Option<u32>,
    verdict: String,
    raw: String,
}

fn parse_plan_steps(text: &str) -> Vec<Step> {
    text.split_whitespace()
        .filter_map(|token| {
            let token = token.trim_matches(|character: char| {
                !character.is_ascii_alphanumeric() && character != '-'
            });
            let parts = token.split('-').collect::<Vec<_>>();
            if parts.len() != 3 {
                return None;
            }
            Some(Step {
                wave: parts[0].strip_prefix(['W', 'w'])?.parse().ok()?,
                lane: parts[1].strip_prefix(['L', 'l'])?.parse().ok()?,
                step: parts[2].strip_prefix(['S', 's'])?.parse().ok()?,
            })
        })
        .collect()
}

fn parse_ledger(text: &str) -> Vec<LedgerRow> {
    text.lines().filter_map(parse_ledger_line).collect()
}
fn parse_ledger_line(line: &str) -> Option<LedgerRow> {
    let fields = line.split_whitespace().collect::<Vec<_>>();
    if fields.len() < 3 || line.trim_start().starts_with('#') {
        return None;
    }
    let lane = fields[0].strip_prefix(['L', 'l'])?.parse().ok()?;
    let scope = fields[1].strip_prefix(['W', 'w'])?;
    let (wave, step) = match scope.split_once(['-', '–']) {
        Some((wave, step)) => (
            wave.parse().ok()?,
            step.strip_prefix(['s', 'S'])?
                .trim_end_matches(|character: char| !character.is_ascii_digit())
                .parse()
                .ok(),
        ),
        None => (scope.parse().ok()?, None),
    };
    let verdict = fields[2].to_ascii_uppercase();
    if !matches!(verdict.as_str(), "PASS" | "REDO" | "FAIL") {
        return None;
    }
    Some(LedgerRow {
        lane,
        wave,
        step,
        verdict,
        raw: line.into(),
    })
}

fn normalized_rows(text: &str) -> Vec<String> {
    text.lines()
        .map(str::trim_end)
        .filter(|line| !line.is_empty() && !line.starts_with('#'))
        .map(str::to_owned)
        .collect()
}
fn now_seconds(context: &ExecutionContext) -> i64 {
    context.now_unix_millis() / 1_000
}
fn empty_dash(value: &str) -> &str {
    if value.is_empty() { "-" } else { value }
}
fn validate_id(kind: &str, value: &str) -> Result<(), CliError> {
    let bytes = value.as_bytes();
    if (1..=64).contains(&bytes.len())
        && (bytes[0].is_ascii_lowercase() || bytes[0].is_ascii_digit())
        && bytes
            .iter()
            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || *byte == b'-')
    {
        Ok(())
    } else {
        usage(format!("unsafe {kind} id `{value}`"))
    }
}
pub(crate) fn is_canonical(value: &str) -> bool {
    canonical_suggestion(value).is_some_and(|candidate| candidate == value)
}
fn canonical_suggestion(value: &str) -> Option<String> {
    let rest = value.strip_prefix('v')?;
    if let Some((version, dev_tail)) = rest.split_once("-dev") {
        let dev = dev_tail
            .bytes()
            .take_while(u8::is_ascii_digit)
            .map(char::from)
            .collect::<String>();
        if !version.is_empty()
            && !dev.is_empty()
            && version.bytes().all(|byte| byte.is_ascii_digit())
        {
            return Some(format!("v{version}-dev{dev}"));
        }
        return None;
    }
    if !rest.is_empty() && rest.bytes().all(|byte| byte.is_ascii_digit()) {
        return Some(format!("v{rest}"));
    }
    None
}
pub(crate) fn derive_id(value: &str, kind: RunKind) -> Result<String, CliError> {
    let version = value
        .strip_prefix('v')
        .ok_or_else(|| usage_error(format!("cannot derive a run id from {value:?}")))?;
    let (numbers, dev) = version.split_once("-dev.").unwrap_or((version, ""));
    let parts = numbers.split('.').collect::<Vec<_>>();
    if parts.len() != 3
        || parts
            .iter()
            .any(|part| part.is_empty() || !part.bytes().all(|byte| byte.is_ascii_digit()))
        || (kind.is_sprint() && (dev.is_empty() || !dev.bytes().all(|byte| byte.is_ascii_digit())))
    {
        return Err(usage_error(format!(
            "cannot derive a run id from {value:?} (expected v{{X}}.{{Y}}.{{Z}} or v{{X}}.{{Y}}.{{Z}}-dev.{{N}})"
        )));
    }
    Ok(if kind.is_sprint() {
        format!("v{}{}{}-dev{dev}", parts[0], parts[1], parts[2])
    } else {
        format!("v{}{}{}", parts[0], parts[1], parts[2])
    })
}
fn output(context: &mut ExecutionContext, value: &str) -> Result<(), CliError> {
    context
        .write_stdout(format!("{value}\n").as_bytes())
        .map_err(|error| CliError::message(format!("write stdout: {error}")))
}
fn error_output(context: &mut ExecutionContext, value: &str) -> Result<(), CliError> {
    context
        .write_stderr(format!("{value}\n").as_bytes())
        .map_err(|error| CliError::message(format!("write stderr: {error}")))
}
fn usage<T>(message: impl Into<String>) -> Result<T, CliError> {
    Err(usage_error(message))
}
fn usage_error(message: impl Into<String>) -> CliError {
    CliError::message_with_code(message, 2)
}
fn missing_or_conflict(message: impl Into<String>) -> Result<(), CliError> {
    Err(CliError::message_with_code(message, 5))
}
fn no_such_run(context: &ExecutionContext, run: &str) -> Result<(), CliError> {
    Err(no_such_run_error(context, run))
}
fn no_such_run_error(context: &ExecutionContext, run: &str) -> CliError {
    CliError::message_with_code(
        format!(
            "no such run: {run} (expected {})",
            context.runs_root.join(run).join("run.json").display()
        ),
        5,
    )
}
fn store_error(error: RunStoreError) -> CliError {
    match error {
        RunStoreError::AlreadyExists(path) => CliError::message_with_code(
            format!(
                "run already exists: {}",
                path.parent()
                    .and_then(Path::file_name)
                    .and_then(|name| name.to_str())
                    .unwrap_or("<unknown>")
            ),
            5,
        ),
        RunStoreError::SchemaAhead(version) => CliError::message_with_code(
            format!("run schema version {version} is newer than this binary supports"),
            2,
        ),
        RunStoreError::Validation(message) | RunStoreError::Mutation(message) => {
            CliError::message_with_code(message, 2)
        }
        other => CliError::message(other.to_string()),
    }
}
fn run_usage() -> &'static str {
    "shepherd run <init|successor|successor-abort|rename|canonicalize|show|list|claim|set|orientation|migrate|lane|layout|ledger|wave>"
}