shepherd-cli 6.6.1

//! Native read-only planning, graph, rendering, and report commands.
//!
//! Every artifact input is anchored below the canonical project namespace.

use std::{
    collections::{BTreeMap, BTreeSet},
    path::{Component, Path, PathBuf},
    process::Command,
};

// Digests come from the engine. This file used to carry its own copy of the
// sha256/hex helper; `scripts/check-rust-duplicates.py` counted fourteen of
// them across cli, render and registry, and asserts one home in shepherd-core.
use shepherd::digest::sha256_hex;
use shepherd::run::{LaneStatus, RunStatus};
use shepherd::{
    RunState,
    dispatch::{LaneId, RunId},
    plan::{
        CommandObservation, InterfaceObservation, PathKind, PathState, PlanEnvironmentProbe,
        PlanProbeManifest, PlanTopology, ProbeExpectation, SourceProbe, VerifiedPlanSeed,
        check_plan_environment, parse_plan, render_lane, render_topology,
        validate_plan_repository_path, validate_plan_structure,
    },
    registry::{OpenMode, Registry},
};

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

const PLAN_USAGE: &str = "usage: shepherd plan <hash|validate|verify|topology|extract|materialize|check> --run <run> [options]";
const GRAPH_USAGE: &str = "usage: shepherd graph <status|diagram|trace> --run <run> [--json]";
const REPORT_USAGE: &str =
    "usage: shepherd report <audit|close|discovery|escalation|teammates|help> [args]";
const MAX_ARTIFACT_BYTES: u64 = 1_048_576;

/// Stable inputs that the native orientation checkpoint binds before execution.
/// Environment observations are rerun, not trusted from this serialized receipt.
#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct PlanReadiness {
    schema: String,
    run: String,
    worktree_root: String,
    worktree_identity: String,
    baseline: String,
    probe_count: usize,
    pub(crate) run_artifacts: BTreeMap<String, String>,
    source_artifacts: BTreeMap<String, String>,
}

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

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Subcommand)]
enum PlanAction {
    Hash(PlanRunCmd),
    Validate(PlanRunCmd),
    Verify(PlanRunCmd),
    Topology(PlanRunCmd),
    Extract(PlanRunCmd),
    Materialize(PlanMaterializeCmd),
    Check(PlanCheckCmd),
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Args)]
struct PlanMaterializeCmd {
    #[arg(long)]
    run: String,
    #[arg(long)]
    check: bool,
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Args)]
struct PlanCheckCmd {
    #[arg(long)]
    run: String,
    #[arg(long)]
    worktree: PathBuf,
    #[arg(long)]
    probes: PathBuf,
    #[arg(long)]
    json: bool,
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Args)]
struct PlanRunCmd {
    #[arg(long)]
    run: String,
    #[arg(long)]
    json: bool,
}

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

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Subcommand)]
enum GraphAction {
    Status(GraphRunCmd),
    Diagram(GraphRunCmd),
    Trace(GraphRunCmd),
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Args)]
struct GraphRunCmd {
    #[arg(long)]
    run: String,
    #[arg(long)]
    json: bool,
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Args)]
pub struct RenderCmd {
    template: String,
    #[arg(long = "var")]
    variables: Vec<String>,
    #[arg(long)]
    vars_json: Option<String>,
    #[arg(long)]
    json: bool,
}

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

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Subcommand)]
enum ReportAction {
    Audit(ReportRunCmd),
    Close(ReportRunCmd),
    Discovery(ReportRunCmd),
    Escalation(ReportEscalationCmd),
    Teammates(ReportTeammatesCmd),
    Help,
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Args)]
struct ReportRunCmd {
    #[arg(long)]
    run: String,
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Args)]
struct ReportEscalationCmd {
    #[arg(long)]
    open_only: bool,
    #[arg(long)]
    json: bool,
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Args)]
struct ReportTeammatesCmd {
    #[arg(long)]
    team: Option<String>,
    #[arg(long)]
    stale_mins: Option<u64>,
    #[arg(long)]
    json: bool,
}

#[derive(serde::Serialize)]
struct ReportEscalationOpen {
    id: i64,
    role: String,
    phase: String,
    question: String,
    raised_at: i64,
}

#[derive(serde::Serialize)]
struct ReportEscalation {
    id: i64,
    role: String,
    question: String,
    raised_at: i64,
    resolved_at: Option<i64>,
}

#[derive(serde::Serialize)]
struct ReportTeammate {
    teammate_name: String,
    agent_type: String,
    status: String,
    last_seen_at: i64,
}

#[derive(serde::Deserialize)]
struct PlanSeedEnvelope {
    schema: String,
    run: String,
    mesh: String,
    outcomes: Vec<PlanSeedId>,
    deliverables: Vec<PlanSeedId>,
}

#[derive(serde::Deserialize)]
struct PlanSeedId {
    id: String,
}

#[derive(serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct ProbeManifestWire {
    schema: String,
    worktree_identity: String,
    baseline: String,
    probes: Vec<ProbeWire>,
}

#[derive(serde::Deserialize)]
#[serde(tag = "kind", rename_all = "kebab-case", deny_unknown_fields)]
enum ProbeWire {
    Path {
        path: String,
        expectation: ProbeExpectationWire,
        path_kind: PathKindWire,
    },
    Symbol {
        path: String,
        symbol: String,
        expected_matches: usize,
    },
    Interface {
        path: String,
        schema: String,
        version: String,
    },
    Command {
        argv: Vec<String>,
        expected_exit: i32,
        semantic_marker: String,
    },
}

#[derive(
    Clone,
    Copy,
    serde::Deserialize,
    strum::AsRefStr,
    strum::Display,
    strum::EnumCount,
    strum::EnumIs,
    strum::EnumString,
    strum::IntoStaticStr,
    strum::VariantNames,
)]
#[serde(rename_all = "lowercase")]
#[strum(ascii_case_insensitive, serialize_all = "snake_case")]
enum ProbeExpectationWire {
    Modify,
    Create,
}

#[derive(
    Clone,
    Copy,
    serde::Deserialize,
    strum::AsRefStr,
    strum::Display,
    strum::EnumCount,
    strum::EnumIs,
    strum::EnumString,
    strum::IntoStaticStr,
    strum::VariantNames,
)]
#[serde(rename_all = "lowercase")]
#[strum(ascii_case_insensitive, serialize_all = "snake_case")]
enum PathKindWire {
    File,
    Directory,
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Args)]
pub struct AuditCmd {
    #[arg(long)]
    run: String,
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Args)]
pub struct DiscoveryCmd {
    #[arg(long)]
    run: String,
}

/// Close one registered lane through the canonical run-state ledger.
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, clap::Args)]
pub struct CloseLaneCmd {
    #[arg(long)]
    run: String,
    lane: String,
    /// Clean closures complete a lane; partial or failed closures preserve an
    /// explicit error state for the next resume owner.
    #[arg(long, default_value = "clean")]
    status: String,
}

impl PlanCmd {
    pub(crate) fn run(self, globals: CliGlobals) -> Result<(), CliError> {
        let Some(action) = self.action else {
            return Err(CliError::message_with_code(PLAN_USAGE, 2));
        };
        let mut context = context(globals)?;
        match action {
            PlanAction::Hash(command) => plan_hash(&mut context, command),
            PlanAction::Validate(command) => validate_plan(&mut context, command),
            PlanAction::Verify(command) => verify_plan(&mut context, command),
            PlanAction::Topology(command) | PlanAction::Extract(command) => {
                topology(&mut context, command)
            }
            PlanAction::Materialize(command) => materialize_plan(&mut context, command),
            PlanAction::Check(command) => check_plan(&mut context, command),
        }
    }
}

impl GraphCmd {
    pub(crate) fn run(self, globals: CliGlobals) -> Result<(), CliError> {
        let Some(action) = self.action else {
            return Err(CliError::message_with_code(GRAPH_USAGE, 2));
        };
        let mut context = context(globals)?;
        match action {
            GraphAction::Status(command) => graph_status(&mut context, command),
            GraphAction::Diagram(command) => graph_diagram(&mut context, command),
            GraphAction::Trace(command) => graph_trace(&mut context, command),
        }
    }
}

impl RenderCmd {
    pub(crate) fn run(self, globals: CliGlobals) -> Result<(), CliError> {
        let mut context = context(globals)?;
        let template = template_path(&context, &self.template)?;
        let variables = render_variables(self.vars_json.as_deref(), &self.variables)?;
        let source = String::from_utf8(read_regular(&template)?)
            .map_err(|_| CliError::message_with_code("template is not UTF-8", 3))?;
        let text = shepherd::render::env::build()
            .template_from_str(&source)
            .and_then(|compiled| compiled.render(&variables))
            .map_err(|error| CliError::message_with_code(error.to_string(), 4))?;
        if self.json {
            let vars = serde_json::to_vec(&variables).map_err(|error| {
                CliError::message_with_code(format!("cannot encode template variables: {error}"), 4)
            })?;
            return write_json(
                &mut context,
                serde_json::json!({
                    "text": text,
                    "manifest": {
                        "template_sha256": sha256_hex(source.as_bytes()),
                        "vars_sha256": sha256_hex(&vars),
                        "output_sha256": sha256_hex(text.as_bytes()),
                    },
                }),
            );
        }
        write_exact(&mut context, text.as_bytes())
    }
}

impl ReportCmd {
    pub(crate) fn run(self, globals: CliGlobals) -> Result<(), CliError> {
        let mut context = context(globals)?;
        let Some(action) = self.action else {
            return write(&mut context, REPORT_USAGE);
        };
        match action {
            ReportAction::Audit(command) => report_audit(&mut context, &command.run),
            ReportAction::Close(command) => report_close(&mut context, &command.run),
            ReportAction::Discovery(command) => report_discovery(&mut context, &command.run),
            ReportAction::Escalation(command) => report_escalation(&mut context, command),
            ReportAction::Teammates(command) => report_teammates(&mut context, command),
            ReportAction::Help => write(&mut context, REPORT_USAGE),
        }
    }
}

impl AuditCmd {
    pub(crate) fn run(self, globals: CliGlobals) -> Result<(), CliError> {
        let mut context = context(globals)?;
        report_audit(&mut context, &self.run)
    }
}

impl DiscoveryCmd {
    pub(crate) fn run(self, globals: CliGlobals) -> Result<(), CliError> {
        let mut context = context(globals)?;
        report_discovery(&mut context, &self.run)
    }
}

impl CloseLaneCmd {
    pub(crate) fn run(self, globals: CliGlobals) -> Result<(), CliError> {
        if !matches!(self.status.as_str(), "clean" | "partial" | "failed") {
            return Err(CliError::message_with_code(
                "--status must be clean, partial, or failed",
                2,
            ));
        }
        LaneId::new(&self.lane)
            .map_err(|error| CliError::message_with_code(error.to_string(), 2))?;
        let mut context = context(globals)?;
        let root = run_dir(&context, &self.run)?;
        let state_path = root.join("run.json");
        let next_state = if self.status == "clean" {
            LaneStatus::Complete
        } else {
            LaneStatus::Error
        };
        RunStore::new(state_path)
            .update(|state| {
                let lane = state
                    .lanes
                    .iter_mut()
                    .find(|lane| lane.id == self.lane)
                    .ok_or_else(|| {
                        crate::RunStoreError::mutation(format!(
                            "no such lane: {} in run {}",
                            self.lane, self.run
                        ))
                    })?;
                lane.state = next_state.into();
                Ok(())
            })
            .map_err(|error| CliError::message_with_code(error.to_string(), 5))?;
        write(
            &mut context,
            &format!(
                "lane {} closed in {} ({})",
                self.lane, self.run, self.status
            ),
        )
    }
}

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 plan_hash(context: &mut ExecutionContext, command: PlanRunCmd) -> Result<(), CliError> {
    let path = plan_path(context, &command.run)?;
    let digest = sha256_hex(&read_regular(&path)?);
    if command.json {
        return write_json(
            context,
            serde_json::json!({"run": command.run, "path": path, "sha256": digest}),
        );
    }
    write(context, &digest)
}

fn validate_plan(context: &mut ExecutionContext, command: PlanRunCmd) -> Result<(), CliError> {
    let path = plan_path(context, &command.run)?;
    let analysis = analyze_plan(&read_text(&path)?);
    let valid = analysis.errors.is_empty();
    if command.json {
        write_json(
            context,
            serde_json::json!({
                "run": command.run,
                "path": path,
                "headings": analysis.headings,
                "lanes": analysis.lanes,
                "errors": analysis.errors,
                "ok": valid,
            }),
        )?;
    } else if valid {
        write(
            context,
            &format!(
                "OK: {} heading(s), {} declared lane(s)",
                analysis.headings.len(),
                analysis.lanes.len()
            ),
        )?;
    } else {
        write(context, &analysis.errors.join("\n"))?;
    }
    if valid {
        Ok(())
    } else {
        Err(CliError::reported_with_code(6))
    }
}

fn verify_plan(context: &mut ExecutionContext, command: PlanRunCmd) -> Result<(), CliError> {
    RunId::new(&command.run).map_err(|error| CliError::message_with_code(error.to_string(), 2))?;
    let store = RunStore::new(context.runs_root.join(&command.run).join("run.json"));
    let state = store
        .load()
        .map_err(|error| plan_error(error.to_string()))?;
    let (_, readiness) = if state.status.is(RunStatus::Planted) {
        verify_plan_readiness(context, &state)?
    } else {
        crate::orientation::verify_checkpointed_plan(context, &state)?
    };
    let current = store
        .load()
        .map_err(|error| plan_error(error.to_string()))?;
    if current.to_canonical_json() != state.to_canonical_json() {
        return Err(plan_error("run state changed during plan verification"));
    }
    if command.json {
        let mut value =
            serde_json::to_value(readiness).map_err(|error| plan_error(error.to_string()))?;
        value["ok"] = serde_json::Value::Bool(true);
        write_json(context, value)
    } else {
        write(
            context,
            "OK: plan-v2, seed, projections, and fresh environment probes verified",
        )
    }
}

/// Validate current-worktree planning artifacts without loading or locking run
/// state. Lifecycle and dispatch callers must supply their already-held state.
pub(crate) fn verify_plan_readiness(
    context: &ExecutionContext,
    state: &RunState,
) -> Result<(PlanTopology, PlanReadiness), CliError> {
    verify_readiness_snapshot(context, state, None)
}

/// Re-run environmental probes on a seed already accepted by the native
/// planned checkpoint. The planting verifier still only accepts planted runs;
/// opening execution cannot change that contract or manufacture planted state.
pub(crate) fn reverify_plan_readiness(
    context: &ExecutionContext,
    state: &RunState,
    binding: &PlanReadiness,
) -> Result<(PlanTopology, PlanReadiness), CliError> {
    verify_readiness_snapshot(context, state, Some(binding))
}

fn verify_readiness_snapshot(
    context: &ExecutionContext,
    state: &RunState,
    binding: Option<&PlanReadiness>,
) -> Result<(PlanTopology, PlanReadiness), CliError> {
    let bound_topology = binding
        .map(|expected| verify_plan_binding(context, state, expected))
        .transpose()?;
    let run = &state.run;
    let root = workspace_plan_root(context, run)?;
    let directory = descriptor::PlanDirectory::open(&root)?;
    let plan_bytes = directory.read("plan.md")?;
    // Preserve the seed bytes that actually passed the native verifier, rather
    // than reopening a possibly different seed for the checkpoint inventory.
    let (topology, seed_bytes) = if let Some(topology) = bound_topology {
        let seed_bytes = directory.read("seed.md")?;
        let expected = binding.expect("bound topology requires a checkpoint");
        for (path, bytes) in [
            ("plan.md", plan_bytes.as_slice()),
            ("seed.md", seed_bytes.as_slice()),
        ] {
            if expected.run_artifacts.get(path) != Some(&sha256_hex(bytes)) {
                return Err(plan_error(format!(
                    "planning checkpoint changed before re-verification: {path}"
                )));
            }
        }
        (topology, seed_bytes)
    } else {
        parse_plan_and_seed(context, run, &plan_bytes, state)?
    };
    let seed_path = format!(".shepherd/runs/{run}/seed.md");
    let plan_path = format!(".shepherd/runs/{run}/plan.md");
    if state.seed != seed_path || (!state.plan.is_empty() && state.plan != plan_path) {
        return Err(plan_error(
            "run seed/plan pointers do not identify the canonical plan-v2 inputs",
        ));
    }
    let mut run_bytes = BTreeMap::from([
        ("plan.md".to_owned(), plan_bytes),
        ("seed.md".to_owned(), seed_bytes),
    ]);
    for (path, expected) in plan_projection_files(&root, &topology)? {
        let relative = path
            .strip_prefix(&root)
            .map_err(|error| plan_error(error.to_string()))?
            .to_string_lossy()
            .replace('\\', "/");
        let actual = directory.read(&relative)?;
        if actual != expected {
            return Err(plan_error(format!(
                "plan projection is not byte-identical: {relative}"
            )));
        }
        run_bytes.insert(relative, actual);
    }
    for relative in ["mesh.md", "phase0.md", "plan-probes.json"] {
        run_bytes.insert(relative.to_owned(), directory.read(relative)?);
    }
    if topology.mesh != format!(".shepherd/runs/{run}/mesh.md")
        || topology.planning_evidence != format!(".shepherd/runs/{run}/phase0.md")
    {
        return Err(plan_error(
            "plan-v2 must bind this run's canonical mesh and planning evidence",
        ));
    }
    let wire: ProbeManifestWire = serde_json::from_slice(&run_bytes["plan-probes.json"])
        .map_err(|error| plan_error(format!("invalid canonical plan-probes.json: {error}")))?;
    let probes = wire.into_manifest();
    let mut path_probes = BTreeSet::new();
    let mut source_paths = BTreeSet::new();
    for probe in &probes.probes {
        match probe {
            PlanEnvironmentProbe::Path {
                path,
                expectation,
                kind,
            } => {
                if !path_probes.insert(path.as_str()) {
                    return Err(plan_error(format!("duplicate path probe: {path}")));
                }
                if *expectation == ProbeExpectation::Modify && *kind == PathKind::File {
                    source_paths.insert(path.as_str());
                }
            }
            PlanEnvironmentProbe::Symbol { path, .. }
            | PlanEnvironmentProbe::Interface { path, .. } => {
                source_paths.insert(path.as_str());
            }
            PlanEnvironmentProbe::Command { argv, .. } => {
                if argv.first().is_some_and(|program| program == "rg")
                    && let Some(path) = argv.last()
                {
                    source_paths.insert(path.as_str());
                }
            }
        }
    }
    for node in &topology.nodes {
        for owned in &node.owns {
            if !path_probes.contains(owned.as_str()) {
                return Err(plan_error(format!(
                    "canonical probes do not cover owned path `{owned}`"
                )));
            }
        }
    }
    let mut source = CliPlanSource::new(
        context.workspace_root.clone(),
        context.config.spawn.max_parallel as usize,
    )?;
    source.bind_planning_baseline(context, state, &topology, &probes.baseline)?;
    let mut source_artifacts = BTreeMap::new();
    for path in &source_paths {
        let resolved = source.resolve(path).map_err(plan_error)?;
        source_artifacts.insert((*path).to_owned(), sha256_hex(&read_regular(&resolved)?));
    }
    let report = check_plan_environment(&topology, &probes, &source)
        .map_err(|error| plan_error(error.to_string()))?;
    for (relative, expected) in &run_bytes {
        if directory.read(relative)? != *expected {
            return Err(plan_error(format!(
                "planning artifact changed during verification: {relative}"
            )));
        }
    }
    for (path, expected) in &source_artifacts {
        if sha256_hex(&read_regular(&source.resolve(path).map_err(plan_error)?)?) != *expected {
            return Err(plan_error(format!(
                "probe source changed during verification: {path}"
            )));
        }
    }
    if source.baseline().map_err(plan_error)? != report.baseline {
        return Err(plan_error("Git baseline changed during plan verification"));
    }
    directory.validate_identity()?;
    let readiness = PlanReadiness {
        schema: "shepherd.plan-readiness/1".into(),
        run: run.clone(),
        worktree_root: context.workspace_root.display().to_string(),
        worktree_identity: report.worktree_identity,
        baseline: report.baseline,
        probe_count: report.probe_count,
        run_artifacts: run_bytes
            .into_iter()
            .map(|(path, bytes)| (path, sha256_hex(&bytes)))
            .collect(),
        source_artifacts,
    };
    Ok((topology, readiness))
}

/// Reopen the immutable planning inputs authenticated by the native post
/// checkpoint. Product sources may change after execution opens; their old
/// probe snapshots are intentionally not a dispatch-time immutability rule.
pub(crate) fn verify_plan_binding(
    context: &ExecutionContext,
    state: &RunState,
    expected: &PlanReadiness,
) -> Result<PlanTopology, CliError> {
    if expected.schema != "shepherd.plan-readiness/1"
        || expected.run != state.run
        || expected.worktree_root != context.workspace_root.display().to_string()
        || expected.worktree_identity != descriptor::directory_identity(&context.workspace_root)?
        || state.seed != format!(".shepherd/runs/{}/seed.md", state.run)
        || (!state.plan.is_empty() && state.plan != format!(".shepherd/runs/{}/plan.md", state.run))
    {
        return Err(plan_error(
            "planning checkpoint does not bind this run and authoring worktree",
        ));
    }
    let root = workspace_plan_root(context, &state.run)?;
    let directory = descriptor::PlanDirectory::open(&root)?;
    let mut bytes = BTreeMap::new();
    for (path, digest) in &expected.run_artifacts {
        let observed = directory.read(path)?;
        if sha256_hex(&observed) != *digest {
            return Err(plan_error(format!(
                "planning checkpoint artifact changed: {path}"
            )));
        }
        bytes.insert(path.as_str(), observed);
    }
    let input = |name: &str| {
        bytes
            .get(name)
            .map(Vec::as_slice)
            .ok_or_else(|| plan_error(format!("planning checkpoint lacks {name}")))
    };
    let topology = topology_from_verified_seed(&state.run, input("plan.md")?, input("seed.md")?)?;
    let mut paths = BTreeSet::from([
        "plan.md".to_owned(),
        "seed.md".to_owned(),
        "mesh.md".to_owned(),
        "phase0.md".to_owned(),
        "plan-probes.json".to_owned(),
    ]);
    for (path, canonical) in plan_projection_files(&root, &topology)? {
        let relative = path
            .strip_prefix(&root)
            .map_err(|error| plan_error(error.to_string()))?
            .to_string_lossy()
            .replace('\\', "/");
        if input(&relative)? != canonical {
            return Err(plan_error(format!(
                "planning checkpoint projection is stale: {relative}"
            )));
        }
        paths.insert(relative);
    }
    if paths != expected.run_artifacts.keys().cloned().collect() {
        return Err(plan_error(
            "planning checkpoint artifact inventory is incomplete or unexpected",
        ));
    }
    directory.validate_identity()?;
    Ok(topology)
}

fn plan_error(message: impl Into<String>) -> CliError {
    CliError::message_with_code(message, 6)
}

fn topology(context: &mut ExecutionContext, command: PlanRunCmd) -> Result<(), CliError> {
    let path = plan_path(context, &command.run)?;
    let analysis = analyze_plan(&read_text(&path)?);
    let value = serde_json::json!({
        "schema": "shepherd.plan-topology/1",
        "run": command.run,
        "path": path,
        "headings": analysis.headings,
        "lanes": analysis.lanes,
        "errors": analysis.errors,
    });
    if command.json {
        write_json(context, value)
    } else {
        let headings = value["headings"].as_array().map_or(0, Vec::len);
        let lanes = value["lanes"].as_array().map_or(0, Vec::len);
        write(context, &format!("headings: {headings}\nlanes: {lanes}"))
    }
}

fn materialize_plan(
    context: &mut ExecutionContext,
    command: PlanMaterializeCmd,
) -> Result<(), CliError> {
    let topology = load_plan_v2(context, &command.run)?;
    let root = workspace_plan_root(context, &command.run)?;
    let files = plan_projection_files(&root, &topology)?;
    if command.check {
        let mut drift = Vec::new();
        for (path, expected) in &files {
            if !descriptor::regular_exists(path)? {
                drift.push(format!("missing `{}`", path.display()));
                continue;
            }
            let actual = read_regular(path)?;
            if actual != *expected {
                drift.push(format!(
                    "stale `{}`: expected {}, observed {}",
                    path.display(),
                    sha256_hex(expected),
                    sha256_hex(&actual)
                ));
            }
        }
        if !drift.is_empty() {
            return Err(CliError::message_with_code(
                format!(
                    "plan projections are not byte-identical:\n{}",
                    drift.join("\n")
                ),
                6,
            ));
        }
        return write(
            context,
            &format!(
                "OK: {} plan-v2 projection(s) are byte-identical",
                files.len()
            ),
        );
    }

    descriptor::ensure_directory(&root.join("graph"))?;
    descriptor::ensure_directory(&root.join("lanes"))?;
    for lane in &topology.lanes {
        descriptor::ensure_directory(&root.join("lanes").join(&lane.id))?;
    }
    for (path, bytes) in &files {
        descriptor::replace_atomic(path, bytes)?;
    }
    write(
        context,
        &format!(
            "materialized {} plan-v2 projection(s); topology_sha256={}",
            files.len(),
            sha256_hex(&render_topology(&topology))
        ),
    )
}

fn check_plan(context: &mut ExecutionContext, command: PlanCheckCmd) -> Result<(), CliError> {
    let topology = load_plan_v2(context, &command.run)?;
    let mut source =
        CliPlanSource::new(command.worktree, context.config.spawn.max_parallel as usize)?;
    let context_identity = descriptor::directory_identity(&context.workspace_root)?;
    if source.identity != context_identity {
        return Err(CliError::message_with_code(
            "--worktree does not identify the canonical selected checkout",
            6,
        ));
    }
    let probes_path = source.resolve_probe_path(&command.probes)?;
    let probes_bytes = descriptor::read_regular(&probes_path, MAX_ARTIFACT_BYTES)?;
    let wire: ProbeManifestWire = serde_json::from_slice(&probes_bytes).map_err(|error| {
        CliError::message_with_code(format!("invalid plan probe manifest: {error}"), 6)
    })?;
    let manifest = wire.into_manifest();
    let state = RunStore::new(context.runs_root.join(&command.run).join("run.json"))
        .load()
        .map_err(|error| plan_error(error.to_string()))?;
    source.bind_planning_baseline(context, &state, &topology, &manifest.baseline)?;
    let report = check_plan_environment(&topology, &manifest, &source)
        .map_err(|error| CliError::message_with_code(error.to_string(), 6))?;
    let topology_bytes = render_topology(&topology);
    let value = serde_json::json!({
        "schema": "shepherd.plan-check-report/1",
        "run": report.run,
        "topology_sha256": sha256_hex(&topology_bytes),
        "probes_sha256": sha256_hex(&probes_bytes),
        "report": {
            "schema": report.schema,
            "worktree_identity": report.worktree_identity,
            "baseline": report.baseline,
            "probe_count": report.probe_count,
            "available_disk_mib": report.available_disk_mib,
            "model_quota": report.model_quota,
            "host_process_ceiling": report.host_process_ceiling,
            "project_spawn_max_parallel": report.project_spawn_max_parallel,
            "capacity_policy": topology.capacity_policy,
            "effective_process_ceiling": topology.capacity.simultaneous_process_ceiling,
        },
    });
    if command.json {
        write_json(context, value)
    } else {
        write(
            context,
            &format!(
                "OK: {} environment probe(s); topology_sha256={}",
                report.probe_count,
                value["topology_sha256"].as_str().unwrap_or_default()
            ),
        )
    }
}

fn load_plan_v2(context: &ExecutionContext, run: &str) -> Result<PlanTopology, CliError> {
    let plan = plan_path(context, run)?;
    parse_plan_v2(context, run, &read_regular(&plan)?)
}

fn parse_plan_v2(
    context: &ExecutionContext,
    run: &str,
    plan_bytes: &[u8],
) -> Result<PlanTopology, CliError> {
    let state = RunStore::new(context.runs_root.join(run).join("run.json"))
        .load()
        .map_err(|error| plan_error(error.to_string()))?;
    parse_plan_and_seed(context, run, plan_bytes, &state).map(|(topology, _)| topology)
}

fn parse_plan_and_seed(
    context: &ExecutionContext,
    run: &str,
    plan_bytes: &[u8],
    state: &RunState,
) -> Result<(PlanTopology, Vec<u8>), CliError> {
    let run_id =
        RunId::new(run).map_err(|error| CliError::message_with_code(error.to_string(), 2))?;
    // Reject a legacy document before asking any separate seed authority to
    // prove readiness. Historical heading-only inspection is not this gate.
    parse_plan(std::str::from_utf8(plan_bytes).map_err(|_| plan_error("plan is not UTF-8"))?)
        .map_err(|error| plan_error(error.to_string()))?;
    let seed_relative = format!(".shepherd/runs/{run}/seed.md");
    let verified = crate::seed_verifier::verify_persisted_seed_with_state(
        &context.workspace_root,
        &run_id,
        &seed_relative,
        state,
    )?;
    let seed_path = context.workspace_root.join(&seed_relative);
    let seed_bytes = read_regular(&seed_path)?;
    if sha256_hex(&seed_bytes) != verified.sha256 || verified.relative_path != seed_relative {
        return Err(CliError::message_with_code(
            "verified seed bytes changed before plan validation",
            6,
        ));
    }
    let topology = topology_from_verified_seed(run, plan_bytes, &seed_bytes)?;
    Ok((topology, seed_bytes))
}

fn topology_from_verified_seed(
    run: &str,
    plan_bytes: &[u8],
    seed_bytes: &[u8],
) -> Result<PlanTopology, CliError> {
    let document =
        parse_plan(std::str::from_utf8(plan_bytes).map_err(|_| plan_error("plan is not UTF-8"))?)
            .map_err(|error| plan_error(error.to_string()))?;
    let seed_text =
        std::str::from_utf8(seed_bytes).map_err(|_| plan_error("verified seed is not UTF-8"))?;
    let frontmatter = markdown_frontmatter(seed_text)?;
    let envelope: PlanSeedEnvelope = serde_saphyr::from_str(frontmatter).map_err(|error| {
        CliError::message_with_code(
            format!("cannot decode verified seed for plan-v2: {error}"),
            6,
        )
    })?;
    if envelope.schema != "shepherd.seed/2" || envelope.run != run {
        return Err(CliError::message_with_code(
            "verified seed schema/run does not match plan-v2",
            6,
        ));
    }
    let mesh = if envelope.mesh.contains('/') {
        envelope.mesh
    } else {
        format!(".shepherd/runs/{run}/{}", envelope.mesh)
    };
    let seed = VerifiedPlanSeed {
        run: run.to_owned(),
        relative_path: format!(".shepherd/runs/{run}/seed.md"),
        mesh,
        deliverables: envelope
            .deliverables
            .into_iter()
            .map(|deliverable| deliverable.id)
            .collect(),
        outcomes: envelope
            .outcomes
            .into_iter()
            .map(|outcome| outcome.id)
            .collect(),
    };
    validate_plan_structure(&document, &seed)
        .map_err(|error| CliError::message_with_code(error.to_string(), 6))
}

fn markdown_frontmatter(markdown: &str) -> Result<&str, CliError> {
    let body = markdown
        .strip_prefix("---\n")
        .ok_or_else(|| CliError::message_with_code("verified seed has no YAML frontmatter", 6))?;
    let (frontmatter, _) = body
        .split_once("\n---")
        .ok_or_else(|| CliError::message_with_code("verified seed frontmatter is not closed", 6))?;
    Ok(frontmatter)
}

fn plan_projection_files(
    root: &Path,
    topology: &PlanTopology,
) -> Result<Vec<(PathBuf, Vec<u8>)>, CliError> {
    let mut files = vec![(root.join("graph/topology.json"), render_topology(topology))];
    for lane in &topology.lanes {
        let bytes = render_lane(topology, &lane.id)
            .map_err(|error| CliError::message_with_code(error.to_string(), 6))?;
        files.push((root.join("lanes").join(&lane.id).join("plan.md"), bytes));
    }
    files.sort_by(|left, right| left.0.cmp(&right.0));
    Ok(files)
}

impl ProbeManifestWire {
    fn into_manifest(self) -> PlanProbeManifest {
        PlanProbeManifest {
            schema: self.schema,
            worktree_identity: self.worktree_identity,
            baseline: self.baseline,
            probes: self.probes.into_iter().map(ProbeWire::into_probe).collect(),
        }
    }
}

impl ProbeWire {
    fn into_probe(self) -> PlanEnvironmentProbe {
        match self {
            Self::Path {
                path,
                expectation,
                path_kind,
            } => PlanEnvironmentProbe::Path {
                path,
                expectation: match expectation {
                    ProbeExpectationWire::Modify => ProbeExpectation::Modify,
                    ProbeExpectationWire::Create => ProbeExpectation::Create,
                },
                kind: match path_kind {
                    PathKindWire::File => PathKind::File,
                    PathKindWire::Directory => PathKind::Directory,
                },
            },
            Self::Symbol {
                path,
                symbol,
                expected_matches,
            } => PlanEnvironmentProbe::Symbol {
                path,
                symbol,
                expected_matches,
            },
            Self::Interface {
                path,
                schema,
                version,
            } => PlanEnvironmentProbe::Interface {
                path,
                schema,
                version,
            },
            Self::Command {
                argv,
                expected_exit,
                semantic_marker,
            } => PlanEnvironmentProbe::Command {
                argv,
                expected_exit,
                semantic_marker,
            },
        }
    }
}

struct CliPlanSource {
    root: PathBuf,
    identity: String,
    model_quota: usize,
    project_spawn_max_parallel: usize,
    planning_baseline: Option<(String, String)>,
}

impl CliPlanSource {
    fn new(root: PathBuf, project_spawn_max_parallel: usize) -> Result<Self, CliError> {
        if !root.is_absolute() {
            return Err(CliError::message_with_code(
                "--worktree must be an explicit absolute path",
                2,
            ));
        }
        let identity = descriptor::directory_identity(&root)?;
        let model_quota = std::env::var("SHEPHERD_MODEL_QUOTA")
            .map_err(|_| {
                CliError::message_with_code(
                    "SHEPHERD_MODEL_QUOTA is required for plan capacity checking",
                    6,
                )
            })?
            .parse::<usize>()
            .map_err(|_| {
                CliError::message_with_code("SHEPHERD_MODEL_QUOTA must be a positive integer", 6)
            })?;
        if model_quota == 0 {
            return Err(CliError::message_with_code(
                "SHEPHERD_MODEL_QUOTA must be a positive integer",
                6,
            ));
        }
        Ok(Self {
            root,
            identity,
            model_quota,
            project_spawn_max_parallel,
            planning_baseline: None,
        })
    }

    fn current_head(&self) -> Result<String, String> {
        let observation = self.direct_command(&[
            "git".into(),
            "rev-parse".into(),
            "--verify".into(),
            "HEAD".into(),
        ])?;
        let head = observation.stdout.trim();
        if observation.exit != 0 || !exact_commit(head) {
            return Err(format!(
                "cannot read exact current Git HEAD: {}",
                observation.stderr
            ));
        }
        Ok(head.into())
    }

    fn bind_planning_baseline(
        &mut self,
        context: &ExecutionContext,
        state: &RunState,
        topology: &PlanTopology,
        baseline: &str,
    ) -> Result<(), CliError> {
        if !exact_commit(baseline) {
            return Err(plan_error(
                "planning baseline must be an exact lowercase40hex commit",
            ));
        }
        let head = self.current_head().map_err(plan_error)?;
        let kind = self
            .direct_command(&[
                "git".into(),
                "cat-file".into(),
                "-t".into(),
                baseline.into(),
            ])
            .map_err(plan_error)?;
        if kind.exit != 0 || kind.stdout.trim() != "commit" {
            return Err(plan_error("planning baseline is not a known commit"));
        }
        if head != baseline {
            let ancestor = self
                .direct_command(&[
                    "git".into(),
                    "merge-base".into(),
                    "--is-ancestor".into(),
                    baseline.into(),
                    head.clone(),
                ])
                .map_err(plan_error)?;
            if ancestor.exit != 0 {
                return Err(plan_error(
                    "planning baseline is not an ancestor of the current HEAD",
                ));
            }
            let allowed = planning_descendant_paths(context, state, topology)?;
            let range = format!("{baseline}..{head}");
            let commits = self
                .direct_command(&[
                    "git".into(),
                    "rev-list".into(),
                    "--max-count=257".into(),
                    range,
                ])
                .map_err(plan_error)?;
            if commits.exit != 0 {
                return Err(plan_error("cannot inspect planning descendant commits"));
            }
            let commits = commits.stdout.lines().collect::<Vec<_>>();
            if commits.is_empty()
                || commits.len() > 256
                || commits.iter().any(|commit| !exact_commit(commit))
            {
                return Err(plan_error(
                    "planning descendant history is absent, invalid or exceeds256 commits",
                ));
            }
            // Inspect every intervening commit, not just the final tree diff:
            // a product change followed by a revert is not planning-only history.
            for commit in commits {
                let delta = self
                    .direct_command(&[
                        "git".into(),
                        "diff-tree".into(),
                        "--root".into(),
                        "--no-commit-id".into(),
                        "--name-only".into(),
                        "--no-renames".into(),
                        "--no-ext-diff".into(),
                        "-r".into(),
                        "-m".into(),
                        "-z".into(),
                        commit.into(),
                        "--".into(),
                    ])
                    .map_err(plan_error)?;
                if delta.exit != 0
                    || u64::try_from(delta.stdout.len())
                        .map_or(true, |length| length > MAX_ARTIFACT_BYTES)
                {
                    return Err(plan_error(
                        "cannot inspect bounded planning descendant delta",
                    ));
                }
                for path in delta.stdout.split('\0').filter(|path| !path.is_empty()) {
                    validate_plan_repository_path(path).map_err(plan_error)?;
                    if !allowed.contains(path) {
                        return Err(plan_error(format!(
                            "planning descendant changes non-planning path: {path}"
                        )));
                    }
                }
            }
        }
        if self.current_head().map_err(plan_error)? != head {
            return Err(plan_error(
                "Git HEAD changed during planning descendant verification",
            ));
        }
        self.planning_baseline = Some((baseline.into(), head));
        Ok(())
    }

    fn resolve(&self, relative: &str) -> Result<PathBuf, String> {
        validate_plan_repository_path(relative).map_err(str::to_owned)?;
        Ok(self.root.join(relative))
    }

    fn resolve_probe_path(&self, requested: &Path) -> Result<PathBuf, CliError> {
        let relative = if requested.is_absolute() {
            requested.strip_prefix(&self.root).map_err(|_| {
                CliError::message_with_code("--probes must stay inside the explicit worktree", 2)
            })?
        } else {
            requested
        };
        let mut components = Vec::new();
        for component in relative.components() {
            let Component::Normal(component) = component else {
                return Err(CliError::message_with_code(
                    "--probes must be a normalized repository-relative path",
                    2,
                ));
            };
            components.push(component.to_str().ok_or_else(|| {
                CliError::message_with_code("--probes path must be valid UTF-8", 2)
            })?);
        }
        let portable = components.join("/");
        validate_plan_repository_path(&portable).map_err(|message| {
            CliError::message_with_code(format!("invalid --probes path: {message}"), 2)
        })?;
        Ok(self.root.join(relative))
    }

    fn direct_command(&self, argv: &[String]) -> Result<CommandObservation, String> {
        let executable = resolve_probe_executable(&self.root, &argv[0])?;
        let mut command = Command::new(executable);
        command
            .args(&argv[1..])
            .current_dir(&self.root)
            .env_clear()
            .env("GIT_NO_REPLACE_OBJECTS", "1");
        for key in ["SystemRoot", "SYSTEMROOT"] {
            if let Some(value) = std::env::var_os(key) {
                command.env(key, value);
            }
        }
        let output = command
            .output()
            .map_err(|error| format!("cannot execute direct probe argv: {error}"))?;
        Ok(CommandObservation {
            exit: output.status.code().unwrap_or(-1),
            stdout: String::from_utf8(output.stdout)
                .map_err(|_| "probe stdout is not UTF-8".to_owned())?,
            stderr: String::from_utf8(output.stderr)
                .map_err(|_| "probe stderr is not UTF-8".to_owned())?,
        })
    }
}

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

/// Original source baseline stays immutable. The execution transition records
/// this separate, actual current head after all current probes have passed.
pub(crate) fn execution_head(context: &ExecutionContext) -> Result<String, CliError> {
    CliPlanSource::new(
        context.workspace_root.clone(),
        context.config.spawn.max_parallel as usize,
    )?
    .current_head()
    .map_err(plan_error)
}

fn planning_descendant_paths(
    context: &ExecutionContext,
    state: &RunState,
    topology: &PlanTopology,
) -> Result<BTreeSet<String>, CliError> {
    let run = &state.run;
    let prefix = format!(".shepherd/runs/{run}/");
    let mut paths = [
        "plan.md",
        "phase0.md",
        "plan-probes.json",
        "graph/topology.json",
        "orientation-manifest.json",
        "orientation-pre.json",
        "orientation-post.json",
    ]
    .into_iter()
    .map(|relative| format!("{prefix}{relative}"))
    .collect::<BTreeSet<_>>();
    for lane in &topology.lanes {
        paths.insert(format!("{prefix}lanes/{}/plan.md", lane.id));
    }
    let authority = crate::native_authority::validate_state(context, run, state, None)?;
    let directory = descriptor::PlanDirectory::open(&context.runs_root.join(run))?;
    let artifacts = authority
        .pre
        .as_ref()
        .map(|pre| ("orientation-pre.json", &pre.pre_sha256))
        .into_iter()
        .chain(
            authority
                .post
                .as_ref()
                .map(|post| ("orientation-post.json", &post.post_sha256)),
        );
    for (artifact, expected) in artifacts {
        let bytes = directory.read(artifact)?;
        if sha256_hex(&bytes) != *expected {
            return Err(plan_error(
                "native orientation hash changed while resolving planning-only descendants",
            ));
        }
        let value: serde_json::Value =
            serde_json::from_slice(&bytes).map_err(|error| plan_error(error.to_string()))?;
        let sources = value["sources"]
            .as_array()
            .ok_or_else(|| plan_error("native orientation lacks source inventory"))?;
        for source in sources {
            let relative = source["path"]
                .as_str()
                .ok_or_else(|| plan_error("native orientation source lacks path"))?;
            validate_plan_repository_path(relative).map_err(plan_error)?;
            if relative == "seed.md"
                || relative == "mesh.md"
                || relative == "run.json"
                || relative.starts_with("dispatch/")
            {
                return Err(plan_error(
                    "native orientation source cannot grant planning delta authority",
                ));
            }
            paths.insert(format!("{prefix}{relative}"));
        }
    }
    directory.validate_identity()?;
    Ok(paths)
}

fn resolve_probe_executable(root: &Path, program: &str) -> Result<PathBuf, String> {
    if program == "git" {
        return crate::dispatch_service::trusted_git_executable()
            .map_err(|error| error.to_string());
    }
    if program != "rg" {
        return Err(format!("probe executable `{program}` is not allowlisted"));
    }
    let canonical_root = std::fs::canonicalize(root)
        .map_err(|error| format!("cannot canonicalize probe worktree: {error}"))?;
    let path = std::env::var_os("PATH").ok_or_else(|| "probe PATH is absent".to_owned())?;
    let executable_names = if cfg!(windows) {
        vec![format!("{program}.exe")]
    } else {
        vec![program.to_owned()]
    };
    for directory in std::env::split_paths(&path) {
        if !directory.is_absolute() {
            continue;
        }
        let Ok(directory) = std::fs::canonicalize(directory) else {
            continue;
        };
        if directory.starts_with(&canonical_root) {
            continue;
        }
        for name in &executable_names {
            let Ok(candidate) = std::fs::canonicalize(directory.join(name)) else {
                continue;
            };
            if !candidate.starts_with(&canonical_root) && is_executable_file(&candidate) {
                return Ok(candidate);
            }
        }
    }
    Err(format!(
        "probe executable `{program}` was not found on an absolute PATH outside the selected worktree"
    ))
}

fn is_executable_file(path: &Path) -> bool {
    let Ok(metadata) = std::fs::metadata(path) else {
        return false;
    };
    if !metadata.is_file() {
        return false;
    }
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        metadata.permissions().mode() & 0o111 != 0
    }
    #[cfg(not(unix))]
    {
        true
    }
}

impl SourceProbe for CliPlanSource {
    type Error = String;

    fn worktree_identity(&self) -> Result<String, Self::Error> {
        descriptor::directory_identity(&self.root).map_err(|error| format!("{error:?}"))
    }

    fn baseline(&self) -> Result<String, Self::Error> {
        let head = self.current_head()?;
        if let Some((baseline, verified_head)) = &self.planning_baseline {
            if head != *verified_head {
                return Err("Git HEAD changed after planning descendant verification".into());
            }
            return Ok(baseline.clone());
        }
        Ok(head)
    }

    fn path_state(&self, path: &str) -> Result<PathState, Self::Error> {
        let path = self.resolve(path)?;
        descriptor::path_state(&path).map_err(|error| format!("{error:?}"))
    }

    fn symbol_matches(&self, path: &str, symbol: &str) -> Result<usize, Self::Error> {
        let path = self.resolve(path)?;
        let bytes = descriptor::read_regular(&path, MAX_ARTIFACT_BYTES)
            .map_err(|error| format!("{error:?}"))?;
        let text = String::from_utf8(bytes).map_err(|_| "symbol source is not UTF-8".to_owned())?;
        Ok(text.match_indices(symbol).count())
    }

    fn interface(&self, path: &str) -> Result<InterfaceObservation, Self::Error> {
        let path = self.resolve(path)?;
        let bytes = descriptor::read_regular(&path, MAX_ARTIFACT_BYTES)
            .map_err(|error| format!("{error:?}"))?;
        let value: serde_json::Value = serde_json::from_slice(&bytes)
            .map_err(|error| format!("invalid interface JSON: {error}"))?;
        Ok(InterfaceObservation {
            schema: value
                .get("schema")
                .and_then(serde_json::Value::as_str)
                .ok_or_else(|| "interface has no string schema".to_owned())?
                .to_owned(),
            version: value
                .get("version")
                .and_then(serde_json::Value::as_str)
                .ok_or_else(|| "interface has no string version".to_owned())?
                .to_owned(),
        })
    }

    fn run(&self, argv: &[String]) -> Result<CommandObservation, Self::Error> {
        self.direct_command(argv)
    }

    fn available_disk_mib(&self) -> Result<u64, Self::Error> {
        descriptor::available_disk_mib(&self.root).map_err(|error| format!("{error:?}"))
    }

    fn model_quota(&self) -> Result<usize, Self::Error> {
        Ok(self.model_quota)
    }

    fn host_process_ceiling(&self) -> Result<usize, Self::Error> {
        std::thread::available_parallelism()
            .map(usize::from)
            .map_err(|error| format!("cannot measure host process ceiling: {error}"))
    }

    fn project_spawn_max_parallel(&self) -> Result<usize, Self::Error> {
        Ok(self.project_spawn_max_parallel)
    }
}

fn graph_status(context: &mut ExecutionContext, command: GraphRunCmd) -> Result<(), CliError> {
    let value = read_json(&graph_path(context, &command.run, "state.json")?)?;
    let nodes = value
        .get("nodes")
        .and_then(serde_json::Value::as_array)
        .map_or(0, Vec::len);
    let edges = value
        .get("edges")
        .and_then(serde_json::Value::as_array)
        .map_or(0, Vec::len);
    if command.json {
        write_json(
            context,
            serde_json::json!({"run": command.run, "nodes": nodes, "edges": edges, "state": value}),
        )
    } else {
        write(
            context,
            &format!("run: {}\nnodes: {nodes}\nedges: {edges}", command.run),
        )
    }
}

fn graph_diagram(context: &mut ExecutionContext, command: GraphRunCmd) -> Result<(), CliError> {
    let value = read_json(&graph_path(context, &command.run, "state.json")?)?;
    let nodes = value
        .get("nodes")
        .and_then(serde_json::Value::as_array)
        .ok_or_else(|| CliError::message_with_code("graph state has no nodes array", 6))?;
    let edges = value
        .get("edges")
        .and_then(serde_json::Value::as_array)
        .ok_or_else(|| CliError::message_with_code("graph state has no edges array", 6))?;
    let mut output = String::from("flowchart TD\n");
    for node in nodes {
        let id = node
            .get("id")
            .and_then(serde_json::Value::as_str)
            .ok_or_else(|| CliError::message_with_code("graph node has no string id", 6))?;
        let label = node
            .get("label")
            .and_then(serde_json::Value::as_str)
            .unwrap_or(id);
        output.push_str(&format!(
            "  {}[\"{}\"]\n",
            mermaid_id(id)?,
            label.replace('"', "'")
        ));
    }
    for edge in edges {
        let from = edge
            .get("from")
            .and_then(serde_json::Value::as_str)
            .ok_or_else(|| CliError::message_with_code("graph edge has no string from", 6))?;
        let to = edge
            .get("to")
            .and_then(serde_json::Value::as_str)
            .ok_or_else(|| CliError::message_with_code("graph edge has no string to", 6))?;
        output.push_str(&format!(
            "  {} --> {}\n",
            mermaid_id(from)?,
            mermaid_id(to)?
        ));
    }
    if command.json {
        write_json(
            context,
            serde_json::json!({"run": command.run, "mermaid": output}),
        )
    } else {
        write_exact(context, output.as_bytes())
    }
}

fn graph_trace(context: &mut ExecutionContext, command: GraphRunCmd) -> Result<(), CliError> {
    let bytes = read_regular(&graph_path(context, &command.run, "trace.jsonl")?)?;
    if command.json {
        let records = String::from_utf8(bytes)
            .map_err(|_| CliError::message_with_code("graph trace is not UTF-8", 6))?
            .lines()
            .filter(|line| !line.trim().is_empty())
            .map(serde_json::from_str::<serde_json::Value>)
            .collect::<Result<Vec<_>, _>>()
            .map_err(|error| {
                CliError::message_with_code(format!("invalid graph trace JSONL: {error}"), 6)
            })?;
        return write_json(
            context,
            serde_json::json!({"run": command.run, "records": records}),
        );
    }
    write_exact(context, &bytes)
}

fn report_audit(context: &mut ExecutionContext, run: &str) -> Result<(), CliError> {
    let files = markdown_files(&run_dir(context, run)?.join("audits"), "")?;
    render_report(context, run, "Audit", files)
}

fn report_discovery(context: &mut ExecutionContext, run: &str) -> Result<(), CliError> {
    let files = markdown_files(&run_dir(context, run)?.join("reports"), "discovery")?;
    render_report(context, run, "Discovery", files)
}

fn report_close(context: &mut ExecutionContext, run: &str) -> Result<(), CliError> {
    let root = run_dir(context, run)?;
    let audits = markdown_files(&root.join("audits"), "")?;
    let discovery = markdown_files(&root.join("reports"), "discovery")?;
    let mut output = format!(
        "# Close report: {run}\n\n## Audit artifacts\n\n{}\n\n## Discovery artifacts\n\n{}\n",
        audits.len(),
        discovery.len()
    );
    let close = root.join("close.md");
    if descriptor::regular_exists(&close)? {
        output.push_str("\n## Close artifact\n\n");
        output.push_str(&read_text(&close)?);
    }
    write_exact(context, output.as_bytes())
}

fn report_escalation(
    context: &mut ExecutionContext,
    command: ReportEscalationCmd,
) -> Result<(), CliError> {
    let registry = Registry::open(&context.registry_path, OpenMode::ReadOnly).map_err(|error| {
        CliError::message_with_code(format!("cannot open canonical registry: {error}"), 5)
    })?;
    let project = report_project_id(&registry)?;
    let rows = if command.open_only {
        registry.query(
            "SELECT id, role, COALESCE(phase, ''), question, raised_at, NULL FROM escalations WHERE project_id = ?1 AND resolved_at IS NULL ORDER BY raised_at ASC",
            rusqlite::params![project],
            |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?, row.get::<_, String>(3)?, row.get::<_, i64>(4)?, row.get::<_, Option<i64>>(5)?)),
        )
    } else {
        registry.query(
            "SELECT id, role, '', question, raised_at, resolved_at FROM escalations WHERE project_id = ?1 ORDER BY raised_at DESC",
            rusqlite::params![project],
            |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?, row.get::<_, String>(3)?, row.get::<_, i64>(4)?, row.get::<_, Option<i64>>(5)?)),
        )
    }.map_err(|error| CliError::message_with_code(format!("cannot query escalations: {error}"), 5))?;
    if command.json {
        if command.open_only {
            let values = rows
                .into_iter()
                .map(
                    |(id, role, phase, question, raised_at, _)| ReportEscalationOpen {
                        id,
                        role,
                        phase,
                        question,
                        raised_at,
                    },
                )
                .collect::<Vec<_>>();
            return write_serialized(context, &values);
        }
        let values = rows
            .into_iter()
            .map(
                |(id, role, _, question, raised_at, resolved_at)| ReportEscalation {
                    id,
                    role,
                    question,
                    raised_at,
                    resolved_at,
                },
            )
            .collect::<Vec<_>>();
        return write_serialized(context, &values);
    }
    let mut output = String::from("# Escalations\n");
    for (id, role, phase, question, raised_at, resolved_at) in rows {
        if command.open_only {
            let phase = if phase.is_empty() { "?" } else { &phase };
            output.push_str(&format!(
                "\n- **#{id} [{role}/{phase}]** {question} (raised: {raised_at})"
            ));
        } else {
            let status = if resolved_at.is_some() {
                "RESOLVED"
            } else {
                "OPEN"
            };
            output.push_str(&format!("\n- **#{id} [{role}/{status}]** {question}"));
        }
    }
    write(context, &output)
}

fn report_teammates(
    context: &mut ExecutionContext,
    command: ReportTeammatesCmd,
) -> Result<(), CliError> {
    // `--stale-mins` was parsed but ignored by the legacy command. Keep that
    // surface stable while intentionally leaving liveness policy to `teammate`.
    let _ = command.stale_mins;
    let registry = Registry::open(&context.registry_path, OpenMode::ReadOnly).map_err(|error| {
        CliError::message_with_code(format!("cannot open canonical registry: {error}"), 5)
    })?;
    let project = report_project_id(&registry)?;
    let rows = registry.query(
        "SELECT teammate_name, agent_type, status, last_seen_at FROM teammates WHERE project_id = ?1 AND (?2 IS NULL OR team_name = ?2) ORDER BY spawned_at DESC",
        rusqlite::params![project, command.team],
        |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?, row.get::<_, i64>(3)?)),
    ).map_err(|error| CliError::message_with_code(format!("cannot query teammates: {error}"), 5))?;
    if command.json {
        let values = rows
            .into_iter()
            .map(
                |(teammate_name, agent_type, status, last_seen_at)| ReportTeammate {
                    teammate_name,
                    agent_type,
                    status,
                    last_seen_at,
                },
            )
            .collect::<Vec<_>>();
        return write_serialized(context, &values);
    }
    let mut output = String::from("# Teammates\n");
    for (name, agent_type, status, last_seen_at) in rows {
        output.push_str(&format!(
            "\n- **{name}** ({agent_type}) — status: {status} — last seen: {last_seen_at}"
        ));
    }
    write(context, &output)
}

fn report_project_id(registry: &Registry) -> Result<String, CliError> {
    registry
        .query("SELECT id FROM projects ORDER BY id LIMIT 1", [], |row| {
            row.get::<_, String>(0)
        })
        .map_err(|error| {
            CliError::message_with_code(format!("cannot query project identity: {error}"), 5)
        })?
        .into_iter()
        .next()
        .ok_or_else(|| {
            CliError::message_with_code("no project registered in the canonical registry", 5)
        })
}

fn render_report(
    context: &mut ExecutionContext,
    run: &str,
    kind: &str,
    files: Vec<PathBuf>,
) -> Result<(), CliError> {
    if files.is_empty() {
        return write(
            context,
            &format!("# {kind} report: {run}\n\n(no {kind} artifacts)"),
        );
    }
    let mut output = format!("# {kind} report: {run}\n");
    for path in files {
        output.push_str("\n---\n\n");
        output.push_str(&read_text(&path)?);
        if !output.ends_with('\n') {
            output.push('\n');
        }
    }
    write_exact(context, output.as_bytes())
}

fn plan_path(context: &ExecutionContext, run: &str) -> Result<PathBuf, CliError> {
    let root = workspace_plan_root(context, run)?;
    RunStore::new(context.runs_root.join(run).join("run.json"))
        .load()
        .map_err(|error| {
            CliError::message_with_code(format!("cannot load requested run: {error}"), 5)
        })?;
    let plan = root.join("plan.md");
    if !descriptor::regular_exists(&plan)? {
        return Err(CliError::message_with_code(
            format!("no canonical plan at {}", plan.display()),
            5,
        ));
    }
    Ok(plan)
}

fn workspace_plan_root(context: &ExecutionContext, run: &str) -> Result<PathBuf, CliError> {
    RunId::new(run).map_err(|error| CliError::message_with_code(error.to_string(), 2))?;
    let root = context.workspace_root.join(".shepherd/runs").join(run);
    if !descriptor::directory_exists(&root)? {
        return Err(plan_error(format!(
            "no planning artifacts in the selected worktree: {}",
            root.display()
        )));
    }
    Ok(root)
}

fn graph_path(context: &ExecutionContext, run: &str, leaf: &str) -> Result<PathBuf, CliError> {
    let path = run_dir(context, run)?.join("graph").join(leaf);
    if !descriptor::regular_exists(&path)? {
        return Err(CliError::message_with_code(
            format!("no graph artifact at {}", path.display()),
            5,
        ));
    }
    Ok(path)
}

fn run_dir(context: &ExecutionContext, run: &str) -> Result<PathBuf, CliError> {
    RunId::new(run).map_err(|error| CliError::message_with_code(error.to_string(), 2))?;
    let path = context.runs_root.join(run);
    if !descriptor::directory_exists(&path)? {
        return Err(CliError::message_with_code(
            format!("no canonical run directory at {}", path.display()),
            5,
        ));
    }
    RunStore::new(path.join("run.json"))
        .load()
        .map_err(|error| {
            CliError::message_with_code(format!("cannot load requested run: {error}"), 5)
        })?;
    Ok(path)
}

fn template_path(context: &ExecutionContext, requested: &str) -> Result<PathBuf, CliError> {
    let relative = safe_relative_path(requested)?;
    let root = context.namespace.join("templates");
    let direct = root.join(&relative);
    let with_extension = if direct.extension().is_none() {
        direct.with_extension("j2")
    } else {
        direct.clone()
    };
    if descriptor::regular_exists(&direct)? {
        return Ok(direct);
    }
    if descriptor::regular_exists(&with_extension)? {
        return Ok(with_extension);
    }
    Err(CliError::message_with_code(
        format!("template not found: {requested}"),
        3,
    ))
}

fn safe_relative_path(value: &str) -> Result<PathBuf, CliError> {
    let path = Path::new(value);
    if value.is_empty()
        || path.is_absolute()
        || path
            .components()
            .any(|part| !matches!(part, Component::Normal(_)))
    {
        return Err(CliError::message_with_code(
            "template path must be a non-empty safe relative path",
            2,
        ));
    }
    Ok(path.to_path_buf())
}

fn render_variables(
    vars_json: Option<&str>,
    pairs: &[String],
) -> Result<serde_json::Value, CliError> {
    let mut value = match vars_json {
        Some(raw) => serde_json::from_str(raw).map_err(|error| {
            CliError::message_with_code(format!("--vars-json must be valid JSON: {error}"), 2)
        })?,
        None => serde_json::Value::Object(serde_json::Map::new()),
    };
    let object = value.as_object_mut().ok_or_else(|| {
        CliError::message_with_code("--vars-json must decode to a JSON object", 2)
    })?;
    for pair in pairs {
        let (key, text) = pair.split_once('=').ok_or_else(|| {
            CliError::message_with_code(format!("--var expects key=value, got: {pair:?}"), 2)
        })?;
        if key.is_empty() {
            return Err(CliError::message_with_code(
                "--var key must be non-empty",
                2,
            ));
        }
        object.insert(key.to_owned(), serde_json::Value::String(text.to_owned()));
    }
    Ok(value)
}

pub(crate) fn read_regular(path: &Path) -> Result<Vec<u8>, CliError> {
    descriptor::read_regular(path, MAX_ARTIFACT_BYTES)
}

fn read_text(path: &Path) -> Result<String, CliError> {
    String::from_utf8(read_regular(path)?)
        .map_err(|_| CliError::message_with_code("artifact is not UTF-8", 6))
}

fn read_json(path: &Path) -> Result<serde_json::Value, CliError> {
    serde_json::from_slice(&read_regular(path)?)
        .map_err(|error| CliError::message_with_code(format!("invalid JSON: {error}"), 6))
}

fn markdown_files(root: &Path, needle: &str) -> Result<Vec<PathBuf>, CliError> {
    let mut files = descriptor::regular_children(root)?
        .into_iter()
        .filter(|name| name.ends_with(".md") && name.contains(needle))
        .map(|name| root.join(name))
        .collect::<Vec<_>>();
    files.sort();
    Ok(files)
}

fn mermaid_id(value: &str) -> Result<String, CliError> {
    if value.is_empty()
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_')
    {
        return Err(CliError::message_with_code(
            format!("graph identifier is not Mermaid-safe: {value:?}"),
            6,
        ));
    }
    Ok(value.to_owned())
}

#[derive(Debug)]
struct PlanAnalysis {
    headings: Vec<String>,
    lanes: Vec<String>,
    errors: Vec<String>,
}

fn analyze_plan(text: &str) -> PlanAnalysis {
    let headings = text
        .lines()
        .filter_map(|line| line.strip_prefix('#').map(str::trim))
        .filter(|heading| !heading.is_empty())
        .map(ToOwned::to_owned)
        .collect::<Vec<_>>();
    let mut lanes = text.lines().filter_map(declared_lane).collect::<Vec<_>>();
    lanes.sort();
    lanes.dedup();
    let mut errors = Vec::new();
    if headings.is_empty() {
        errors.push("ERROR: plan has no Markdown heading".into());
    }
    if text.contains('\0') {
        errors.push("ERROR: plan contains NUL bytes".into());
    }
    PlanAnalysis {
        headings,
        lanes,
        errors,
    }
}

fn declared_lane(line: &str) -> Option<String> {
    let marker = line.trim().strip_prefix("lane:")?.trim();
    RunId::new(marker).ok().map(|id| id.as_str().to_owned())
}

fn write(context: &mut ExecutionContext, output: &str) -> Result<(), CliError> {
    write_exact(context, format!("{output}\n").as_bytes())
}

fn write_exact(context: &mut ExecutionContext, bytes: &[u8]) -> Result<(), CliError> {
    context
        .write_stdout(bytes)
        .map_err(|error| CliError::message(format!("cannot write stdout: {error}")))
}

fn write_json(context: &mut ExecutionContext, value: serde_json::Value) -> Result<(), CliError> {
    let output = serde_json::to_string_pretty(&value)
        .map_err(|error| CliError::message(format!("cannot encode JSON: {error}")))?;
    write(context, &output)
}

fn write_serialized<T: serde::Serialize>(
    context: &mut ExecutionContext,
    value: &T,
) -> Result<(), CliError> {
    let output = serde_json::to_string_pretty(value)
        .map_err(|error| CliError::message(format!("cannot encode JSON: {error}")))?;
    write(context, &output)
}

#[cfg(unix)]
mod descriptor {
    use std::{
        fs::File,
        io::{Read, Write},
        os::fd::OwnedFd,
        path::{Component, Path, PathBuf},
        sync::atomic::{AtomicU64, Ordering},
    };

    use rustix::fs::{
        self, AtFlags, Dir, FileType, Mode, OFlags, open, openat, renameat, unlinkat,
    };

    use super::{
        CliError, MAX_ARTIFACT_BYTES, PathState, plan_error, validate_plan_repository_path,
    };

    static PLAN_TEMP_COUNTER: AtomicU64 = AtomicU64::new(0);

    pub(super) struct PlanDirectory {
        root: PathBuf,
        directory: OwnedFd,
        identity: String,
    }

    impl PlanDirectory {
        pub(super) fn open(root: &Path) -> Result<Self, CliError> {
            let directory = open_absolute_directory(root)
                .map_err(|error| plan_error(format!("cannot open planning directory: {error}")))?;
            let stat = fs::fstat(&directory).map_err(|error| plan_error(error.to_string()))?;
            Ok(Self {
                root: root.to_owned(),
                directory,
                identity: format!("unix:{:x}:{:x}", stat.st_dev, stat.st_ino),
            })
        }

        pub(super) fn validate_identity(&self) -> Result<(), CliError> {
            if directory_identity(&self.root)? != self.identity {
                return Err(plan_error(
                    "planning directory was replaced during verification",
                ));
            }
            Ok(())
        }

        pub(super) fn read(&self, relative: &str) -> Result<Vec<u8>, CliError> {
            validate_plan_repository_path(relative).map_err(plan_error)?;
            self.validate_identity()?;
            let mut parent = self
                .directory
                .try_clone()
                .map_err(|error| plan_error(error.to_string()))?;
            let mut parts = Path::new(relative).components().peekable();
            while let Some(component) = parts.next() {
                let Component::Normal(name) = component else {
                    return Err(plan_error("noncanonical planning artifact path"));
                };
                if parts.peek().is_some() {
                    parent = openat(&parent, name, directory_flags(), Mode::empty()).map_err(
                        |error| {
                            plan_error(format!("cannot open planning artifact {relative}: {error}"))
                        },
                    )?;
                    continue;
                }
                let fd = openat(
                    &parent,
                    name,
                    OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW | OFlags::NONBLOCK,
                    Mode::empty(),
                )
                .map_err(|error| {
                    plan_error(format!("cannot open planning artifact {relative}: {error}"))
                })?;
                let file = File::from(fd);
                let before = file
                    .metadata()
                    .map_err(|error| plan_error(error.to_string()))?;
                if !before.is_file() || before.len() > MAX_ARTIFACT_BYTES {
                    return Err(plan_error(format!(
                        "planning artifact is not a bounded regular file: {relative}"
                    )));
                }
                let mut bytes = Vec::new();
                (&file)
                    .take(MAX_ARTIFACT_BYTES + 1)
                    .read_to_end(&mut bytes)
                    .map_err(|error| plan_error(error.to_string()))?;
                let after = file
                    .metadata()
                    .map_err(|error| plan_error(error.to_string()))?;
                if bytes.len() as u64 != before.len()
                    || after.len() != before.len()
                    || after.modified().ok() != before.modified().ok()
                {
                    return Err(plan_error(format!(
                        "planning artifact changed while reading: {relative}"
                    )));
                }
                self.validate_identity()?;
                return Ok(bytes);
            }
            Err(plan_error("empty planning artifact path"))
        }
    }

    fn directory_flags() -> OFlags {
        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW
    }

    fn open_absolute_directory(path: &Path) -> Result<OwnedFd, rustix::io::Errno> {
        if !path.is_absolute() {
            return Err(rustix::io::Errno::INVAL);
        }
        let mut directory = open("/", directory_flags(), Mode::empty())?;
        for component in path.components() {
            match component {
                Component::RootDir | Component::Prefix(_) => {}
                Component::Normal(name) => {
                    directory = openat(&directory, name, directory_flags(), Mode::empty())?;
                }
                Component::CurDir | Component::ParentDir => {
                    return Err(rustix::io::Errno::INVAL);
                }
            }
        }
        Ok(directory)
    }

    fn parent_and_name(path: &Path) -> Result<(OwnedFd, &std::ffi::OsStr), rustix::io::Errno> {
        let parent = path.parent().ok_or(rustix::io::Errno::INVAL)?;
        let name = path.file_name().ok_or(rustix::io::Errno::INVAL)?;
        Ok((open_absolute_directory(parent)?, name))
    }

    fn open_regular(path: &Path) -> Result<Option<File>, CliError> {
        let (parent, name) = match parent_and_name(path) {
            Ok(parts) => parts,
            Err(rustix::io::Errno::NOENT) => {
                return Ok(None);
            }
            Err(error) => {
                return Err(CliError::message_with_code(
                    format!(
                        "cannot open canonical parent for {} without following links: {error}",
                        path.display()
                    ),
                    2,
                ));
            }
        };
        let fd = match openat(
            &parent,
            name,
            OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::empty(),
        ) {
            Ok(fd) => fd,
            Err(rustix::io::Errno::NOENT) => return Ok(None),
            Err(error) => {
                return Err(CliError::message_with_code(
                    format!(
                        "cannot open canonical artifact {} without following links: {error}",
                        path.display()
                    ),
                    2,
                ));
            }
        };
        let stat = fs::fstat(&fd).map_err(|error| {
            CliError::message(format!(
                "cannot inspect canonical artifact {}: {error}",
                path.display()
            ))
        })?;
        if !FileType::from_raw_mode(stat.st_mode).is_file() {
            return Err(CliError::message_with_code(
                format!("artifact is not a regular file: {}", path.display()),
                2,
            ));
        }
        Ok(Some(File::from(fd)))
    }

    pub(super) fn read_regular(path: &Path, limit: u64) -> Result<Vec<u8>, CliError> {
        let Some(file) = open_regular(path)? else {
            return Err(CliError::message_with_code(
                format!("artifact is missing: {}", path.display()),
                5,
            ));
        };
        let mut bytes = Vec::new();
        file.take(limit + 1)
            .read_to_end(&mut bytes)
            .map_err(|error| {
                CliError::message(format!("cannot read {}: {error}", path.display()))
            })?;
        if bytes.len() as u64 > limit {
            return Err(CliError::message_with_code(
                format!("artifact exceeds {MAX_ARTIFACT_BYTES} byte limit"),
                6,
            ));
        }
        Ok(bytes)
    }

    pub(super) fn regular_exists(path: &Path) -> Result<bool, CliError> {
        Ok(open_regular(path)?.is_some())
    }

    pub(super) fn directory_exists(path: &Path) -> Result<bool, CliError> {
        match open_absolute_directory(path) {
            Ok(_) => Ok(true),
            Err(rustix::io::Errno::NOENT) => Ok(false),
            Err(error) => Err(CliError::message_with_code(
                format!(
                    "cannot open canonical directory {} without following links: {error}",
                    path.display()
                ),
                2,
            )),
        }
    }

    pub(super) fn regular_children(root: &Path) -> Result<Vec<String>, CliError> {
        let directory = match open_absolute_directory(root) {
            Ok(directory) => directory,
            Err(rustix::io::Errno::NOENT) => return Ok(Vec::new()),
            Err(error) => {
                return Err(CliError::message_with_code(
                    format!(
                        "cannot open canonical directory {} without following links: {error}",
                        root.display()
                    ),
                    2,
                ));
            }
        };
        let entries = Dir::read_from(&directory).map_err(|error| {
            CliError::message(format!("cannot enumerate {}: {error}", root.display()))
        })?;
        let mut names = Vec::new();
        for entry in entries {
            let entry = entry.map_err(|error| {
                CliError::message(format!("cannot enumerate {}: {error}", root.display()))
            })?;
            let name = entry.file_name();
            if name.to_bytes() == b"." || name.to_bytes() == b".." {
                continue;
            }
            let Some(name) = name.to_str().ok() else {
                continue;
            };
            let stat =
                fs::statat(&directory, name, AtFlags::SYMLINK_NOFOLLOW).map_err(|error| {
                    CliError::message(format!(
                        "cannot inspect {}: {error}",
                        root.join(name).display()
                    ))
                })?;
            if FileType::from_raw_mode(stat.st_mode).is_symlink() {
                return Err(CliError::message_with_code(
                    format!(
                        "refusing symlinked planning artifact: {}",
                        root.join(name).display()
                    ),
                    2,
                ));
            }
            if FileType::from_raw_mode(stat.st_mode).is_file() {
                names.push(name.to_owned());
            }
        }
        Ok(names)
    }

    pub(crate) fn directory_identity(path: &Path) -> Result<String, CliError> {
        let directory = open_absolute_directory(path).map_err(|error| {
            CliError::message_with_code(
                format!(
                    "cannot open worktree {} without following links: {error}",
                    path.display()
                ),
                2,
            )
        })?;
        let stat = fs::fstat(&directory).map_err(|error| {
            CliError::message(format!(
                "cannot inspect worktree {}: {error}",
                path.display()
            ))
        })?;
        Ok(format!("unix:{:x}:{:x}", stat.st_dev, stat.st_ino))
    }

    pub(super) fn path_state(path: &Path) -> Result<PathState, CliError> {
        let (parent, name) = match parent_and_name(path) {
            Ok(parts) => parts,
            Err(rustix::io::Errno::NOENT) => return Ok(PathState::Missing),
            Err(error) => {
                return Err(CliError::message_with_code(
                    format!(
                        "cannot inspect {} without following links: {error}",
                        path.display()
                    ),
                    2,
                ));
            }
        };
        let stat = match fs::statat(&parent, name, AtFlags::SYMLINK_NOFOLLOW) {
            Ok(stat) => stat,
            Err(rustix::io::Errno::NOENT) => return Ok(PathState::Missing),
            Err(error) => {
                return Err(CliError::message(format!(
                    "cannot inspect {}: {error}",
                    path.display()
                )));
            }
        };
        let kind = FileType::from_raw_mode(stat.st_mode);
        Ok(if kind.is_file() {
            PathState::File
        } else if kind.is_dir() {
            PathState::Directory
        } else if kind.is_symlink() {
            PathState::Symlink
        } else {
            PathState::Other
        })
    }

    pub(super) fn ensure_directory(path: &Path) -> Result<(), CliError> {
        if directory_exists(path)? {
            return Ok(());
        }
        let (parent, name) = parent_and_name(path).map_err(|error| {
            CliError::message_with_code(
                format!("cannot anchor directory {}: {error}", path.display()),
                2,
            )
        })?;
        match fs::mkdirat(&parent, name, Mode::from_raw_mode(0o755)) {
            Ok(()) | Err(rustix::io::Errno::EXIST) => {}
            Err(error) => {
                return Err(CliError::message(format!(
                    "cannot create directory {}: {error}",
                    path.display()
                )));
            }
        }
        openat(&parent, name, directory_flags(), Mode::empty()).map_err(|error| {
            CliError::message_with_code(
                format!(
                    "created directory {} cannot be reopened without following links: {error}",
                    path.display()
                ),
                2,
            )
        })?;
        fs::fsync(&parent).map_err(|error| {
            CliError::message(format!(
                "cannot sync directory parent for {}: {error}",
                path.display()
            ))
        })
    }

    pub(super) fn replace_atomic(path: &Path, bytes: &[u8]) -> Result<(), CliError> {
        match path_state(path)? {
            PathState::Symlink => {
                return Err(CliError::message_with_code(
                    format!("refusing symlinked plan projection: {}", path.display()),
                    2,
                ));
            }
            PathState::Directory | PathState::Other => {
                return Err(CliError::message_with_code(
                    format!("plan projection is not a regular file: {}", path.display()),
                    2,
                ));
            }
            PathState::File if read_regular(path, MAX_ARTIFACT_BYTES)? == bytes => return Ok(()),
            PathState::Missing | PathState::File => {}
        }
        let (parent, name) = parent_and_name(path).map_err(|error| {
            CliError::message_with_code(
                format!("cannot anchor plan projection {}: {error}", path.display()),
                2,
            )
        })?;
        let nonce = PLAN_TEMP_COUNTER.fetch_add(1, Ordering::Relaxed);
        let temporary = format!(
            ".{}.shepherd.plan.{}.{}",
            name.to_string_lossy(),
            std::process::id(),
            nonce
        );
        let descriptor = openat(
            &parent,
            temporary.as_str(),
            OFlags::WRONLY | OFlags::CREATE | OFlags::EXCL | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::from_raw_mode(0o644),
        )
        .map_err(|error| {
            CliError::message(format!(
                "cannot create atomic plan projection {}: {error}",
                path.display()
            ))
        })?;
        let mut file = File::from(descriptor);
        let result = (|| {
            file.write_all(bytes).map_err(|error| {
                CliError::message(format!(
                    "cannot write plan projection {}: {error}",
                    path.display()
                ))
            })?;
            file.sync_all().map_err(|error| {
                CliError::message(format!(
                    "cannot sync plan projection {}: {error}",
                    path.display()
                ))
            })?;
            renameat(&parent, temporary.as_str(), &parent, name).map_err(|error| {
                CliError::message(format!(
                    "cannot publish plan projection {} atomically: {error}",
                    path.display()
                ))
            })?;
            fs::fsync(&parent).map_err(|error| {
                CliError::message(format!(
                    "cannot sync plan projection parent {}: {error}",
                    path.display()
                ))
            })
        })();
        if result.is_err() {
            let _ = unlinkat(&parent, temporary.as_str(), AtFlags::empty());
        }
        result
    }

    pub(super) fn available_disk_mib(path: &Path) -> Result<u64, CliError> {
        let stat = fs::statvfs(path).map_err(|error| {
            CliError::message(format!(
                "cannot inspect available disk at {}: {error}",
                path.display()
            ))
        })?;
        Ok(stat.f_bavail.saturating_mul(stat.f_frsize) / (1024 * 1024))
    }
}

#[cfg(not(unix))]
mod descriptor {
    use std::{
        fs::File,
        path::{Path, PathBuf},
    };

    use super::{
        CliError, MAX_ARTIFACT_BYTES, PathState, plan_error, validate_plan_repository_path,
    };
    use crate::safe_fs;

    pub(super) struct PlanDirectory {
        root: PathBuf,
        identity: String,
        _directory: File,
    }

    impl PlanDirectory {
        pub(super) fn open(root: &Path) -> Result<Self, CliError> {
            let identity = directory_identity(root)?;
            #[cfg(windows)]
            let directory = {
                use std::os::windows::fs::OpenOptionsExt;
                std::fs::OpenOptions::new()
                    .read(true)
                    .custom_flags(0x0220_0000)
                    .open(root)
            };
            #[cfg(not(windows))]
            let directory = File::open(root);
            let directory = directory.map_err(|error| {
                plan_error(format!("cannot retain planning directory: {error}"))
            })?;
            let value = Self {
                root: root.to_owned(),
                identity,
                _directory: directory,
            };
            value.validate_identity()?;
            Ok(value)
        }

        pub(super) fn validate_identity(&self) -> Result<(), CliError> {
            if directory_identity(&self.root)? != self.identity {
                return Err(plan_error(
                    "planning directory was replaced during verification",
                ));
            }
            Ok(())
        }

        pub(super) fn read(&self, relative: &str) -> Result<Vec<u8>, CliError> {
            validate_plan_repository_path(relative).map_err(plan_error)?;
            self.validate_identity()?;
            let bytes =
                safe_fs::read_regular_nofollow(&self.root.join(relative), MAX_ARTIFACT_BYTES)
                    .map_err(|error| {
                        plan_error(format!("cannot open planning artifact {relative}: {error}"))
                    })?;
            self.validate_identity()?;
            Ok(bytes)
        }
    }

    /// The exit codes are the contract, not decoration: 2 is "cannot open
    /// safely", 5 is "missing", 6 is "over the limit". The unix twin returns
    /// exactly these, and scripts branch on them.
    pub(super) fn read_regular(path: &Path, limit: u64) -> Result<Vec<u8>, CliError> {
        match safe_fs::read_regular_nofollow(path, limit) {
            Ok(bytes) => Ok(bytes),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Err(
                CliError::message_with_code(format!("artifact is missing: {}", path.display()), 5),
            ),
            Err(error) if error.kind() == std::io::ErrorKind::InvalidData => {
                Err(CliError::message_with_code(
                    format!("artifact exceeds {MAX_ARTIFACT_BYTES} byte limit"),
                    6,
                ))
            }
            Err(error) if error.kind() == std::io::ErrorKind::InvalidInput => {
                Err(CliError::message_with_code(
                    format!("artifact is not a regular file: {}", path.display()),
                    2,
                ))
            }
            Err(error) => Err(CliError::message(format!(
                "cannot read {}: {error}",
                path.display()
            ))),
        }
    }

    pub(super) fn regular_exists(path: &Path) -> Result<bool, CliError> {
        safe_fs::regular_exists(path).map_err(|error| {
            CliError::message_with_code(
                format!(
                    "cannot open canonical artifact {} without following links: {error}",
                    path.display()
                ),
                2,
            )
        })
    }

    pub(super) fn directory_exists(path: &Path) -> Result<bool, CliError> {
        safe_fs::directory_exists(path).map_err(|error| {
            CliError::message_with_code(
                format!(
                    "cannot open canonical directory {} without following links: {error}",
                    path.display()
                ),
                2,
            )
        })
    }

    /// An absent directory is an empty listing, not an error: the unix twin
    /// maps ENOENT to `Vec::new()` and callers rely on that to render "no
    /// artifacts yet" instead of failing a status command.
    pub(super) fn regular_children(root: &Path) -> Result<Vec<String>, CliError> {
        match safe_fs::regular_children(root) {
            Ok(names) => Ok(names),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
            Err(error) => Err(CliError::message_with_code(
                format!(
                    "cannot open canonical directory {} without following links: {error}",
                    root.display()
                ),
                2,
            )),
        }
    }

    pub(crate) fn directory_identity(path: &Path) -> Result<String, CliError> {
        safe_fs::reject_link_components(path).map_err(|error| {
            CliError::message_with_code(
                format!(
                    "cannot open worktree {} without following links: {error}",
                    path.display()
                ),
                2,
            )
        })?;
        let metadata = std::fs::metadata(path).map_err(|error| {
            CliError::message(format!(
                "cannot inspect worktree {}: {error}",
                path.display()
            ))
        })?;
        if !metadata.is_dir() {
            return Err(CliError::message_with_code(
                format!("worktree is not a directory: {}", path.display()),
                2,
            ));
        }
        #[cfg(windows)]
        {
            use std::os::windows::fs::MetadataExt;
            return Ok(format!(
                "windows:{}:{:x}:{:x}",
                std::fs::canonicalize(path)
                    .map_err(|error| CliError::message(error.to_string()))?
                    .display(),
                metadata.creation_time(),
                metadata.last_write_time()
            ));
        }
        #[cfg(not(windows))]
        Ok(format!(
            "path:{}",
            std::fs::canonicalize(path)
                .map_err(|error| CliError::message(error.to_string()))?
                .display()
        ))
    }

    pub(super) fn path_state(path: &Path) -> Result<PathState, CliError> {
        if safe_fs::is_link(path).map_err(|error| CliError::message(error.to_string()))? {
            return Ok(PathState::Symlink);
        }
        safe_fs::reject_link_components(path).map_err(|error| {
            CliError::message_with_code(
                format!(
                    "cannot inspect {} without following links: {error}",
                    path.display()
                ),
                2,
            )
        })?;
        match std::fs::symlink_metadata(path) {
            Ok(metadata) if metadata.is_file() => Ok(PathState::File),
            Ok(metadata) if metadata.is_dir() => Ok(PathState::Directory),
            Ok(_) => Ok(PathState::Other),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(PathState::Missing),
            Err(error) => Err(CliError::message(format!(
                "cannot inspect {}: {error}",
                path.display()
            ))),
        }
    }

    pub(super) fn ensure_directory(path: &Path) -> Result<(), CliError> {
        if directory_exists(path)? {
            return Ok(());
        }
        let parent = path.parent().ok_or_else(|| {
            CliError::message_with_code(format!("directory has no parent: {}", path.display()), 2)
        })?;
        let name = path
            .file_name()
            .and_then(|name| name.to_str())
            .ok_or_else(|| {
                CliError::message_with_code(
                    format!("directory name is not UTF-8: {}", path.display()),
                    2,
                )
            })?;
        safe_fs::ensure_directory(parent, name).map_err(|error| {
            CliError::message_with_code(
                format!("cannot create directory {}: {error}", path.display()),
                2,
            )
        })?;
        Ok(())
    }

    pub(super) fn replace_atomic(path: &Path, bytes: &[u8]) -> Result<(), CliError> {
        safe_fs::replace_atomic(path, bytes).map_err(|error| {
            CliError::message_with_code(
                format!("cannot write plan projection {}: {error}", path.display()),
                if error.kind() == std::io::ErrorKind::InvalidInput {
                    2
                } else {
                    5
                },
            )
        })
    }

    #[cfg(windows)]
    #[allow(unsafe_code)]
    pub(super) fn available_disk_mib(path: &Path) -> Result<u64, CliError> {
        use std::os::windows::ffi::OsStrExt;
        use windows_sys::Win32::Storage::FileSystem::GetDiskFreeSpaceExW;

        let wide = path
            .as_os_str()
            .encode_wide()
            .chain(core::iter::once(0))
            .collect::<Vec<_>>();
        let mut available = 0u64;
        // SAFETY: `wide` is NUL-terminated for the duration of the call and the
        // output points to a live u64. Unneeded outputs are null.
        let ok = unsafe {
            GetDiskFreeSpaceExW(
                wide.as_ptr(),
                &mut available,
                core::ptr::null_mut(),
                core::ptr::null_mut(),
            )
        };
        if ok == 0 {
            return Err(CliError::message(
                std::io::Error::last_os_error().to_string(),
            ));
        }
        Ok(available / (1024 * 1024))
    }

    #[cfg(not(windows))]
    pub(super) fn available_disk_mib(_path: &Path) -> Result<u64, CliError> {
        std::env::var("SHEPHERD_AVAILABLE_DISK_MIB")
            .map_err(|_| CliError::message("SHEPHERD_AVAILABLE_DISK_MIB is required"))?
            .parse()
            .map_err(|_| CliError::message("SHEPHERD_AVAILABLE_DISK_MIB must be an integer"))
    }
}

/// The worktree identity production observes, for tests that must match it.
///
/// A second implementation of this format lived in the integration tests and
/// drifted: it emitted `path:<canonical>` while this crate emits
/// `windows:<canonical>:<ctime>:<index>`, so every plan-verify fixture failed
/// on Windows and nothing on unix noticed. One source of truth, and the
/// format is not duplicated across the crate boundary.
pub fn worktree_identity(path: &std::path::Path) -> Result<String, CliError> {
    descriptor::directory_identity(path)
}

#[cfg(test)]
mod tests {
    use super::{
        analyze_plan, declared_lane, mermaid_id, render_variables, safe_relative_path, sha256_hex,
    };

    #[test]
    fn plan_analysis_is_sorted_and_rejects_headingless_documents() {
        let analysis = analyze_plan("# Sprint\nlane: l2\nlane: l1\nlane: l2\n");
        assert_eq!(analysis.headings, ["Sprint"]);
        assert_eq!(analysis.lanes, ["l1", "l2"]);
        assert!(analysis.errors.is_empty());

        let missing = analyze_plan("lane: l1\n");
        assert_eq!(missing.errors, ["ERROR: plan has no Markdown heading"]);
    }

    #[test]
    fn paths_and_graph_identifiers_fail_closed() {
        assert!(safe_relative_path("nested/template.j2").is_ok());
        assert!(safe_relative_path("../escape.j2").is_err());
        assert!(safe_relative_path("/absolute.j2").is_err());
        assert_eq!(mermaid_id("node_1").expect("safe id"), "node_1");
        assert!(mermaid_id("bad-id").is_err());
        assert_eq!(declared_lane(" lane: l1 "), Some("l1".into()));
    }

    #[test]
    fn variables_override_and_hash_is_deterministic() {
        let value = render_variables(
            Some(r#"{"answer":42,"name":"base"}"#),
            &["name=explicit".into()],
        )
        .expect("valid render variables");
        assert_eq!(value["answer"], 42);
        assert_eq!(value["name"], "explicit");
        assert_eq!(
            sha256_hex(b"abc"),
            "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
        );
    }
}