shepherd-cli 6.7.1

//! Native custody for run incarnations and Planning checkpoints.
//!
//! The registry is deliberately outside `.shepherd/runs/<run>/`. Run files and
//! orientation reports are caller-visible artifacts; this file is the native
//! identity ledger that makes copying either one insufficient.

use std::{
    collections::BTreeMap,
    fs::{self, File, OpenOptions},
    io::{Read, Write},
    path::{Path, PathBuf},
    time::Duration,
};

use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use uuid::Uuid;

use crate::{ExecutionContext, interface::CliError};
use shepherd::RunState;
use shepherd::run::RunStatus;

pub(crate) const REGISTRY_FILE: &str = "native-orientation-registry.json";
const LOCK_FILE: &str = "native-orientation-registry.lock";
const REGISTRY_SCHEMA: &str = "shepherd.native-orientation-registry/1";
const POLICY: &str = "v657-planning";
const MAX_BYTES: u64 = 256 * 1024;
const MAX_ABORTED_SUCCESSORS_PER_RUN: usize = 64;
const SUCCESSOR_TERMINAL_RESERVATION_BYTES: usize = 512;
const LOCK_TIMEOUT: Duration = Duration::from_secs(5);

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct Registry {
    schema: String,
    policy: String,
    project_root: String,
    runs: BTreeMap<String, RunAuthority>,
    /// A durable, resumable intent exists before any run directory moves.
    /// Ordinary run validation remains fail-closed while this map is nonempty.
    #[serde(default)]
    pending_successors: BTreeMap<String, PendingSuccessor>,
    #[serde(default)]
    aborted_successors: BTreeMap<String, Vec<AbortedSuccessor>>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct RunAuthority {
    pub(crate) run: String,
    pub(crate) project_id: Option<String>,
    pub(crate) run_incarnation: String,
    pub(crate) orientation_epoch: u64,
    pub(crate) initialized_at: i64,
    pub(crate) pre: Option<PreCheckpoint>,
    pub(crate) post: Option<PostCheckpoint>,
    #[serde(default)]
    pub(crate) pre_history: Vec<RetiredPreCheckpoint>,
    /// Exact legacy source intent, retained to recover an interrupted adoption.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(crate) legacy_adoption_sha256: Option<String>,
    /// Immutable Native history for every retired incarnation of this
    /// canonical run id. The archived artifact tree, rather than this vector,
    /// owns the historical caller-visible bytes.
    #[serde(default)]
    pub(crate) predecessors: Vec<RetiredRunIncarnation>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct RetiredRunIncarnation {
    pub(crate) run_incarnation: String,
    pub(crate) orientation_epoch: u64,
    pub(crate) initialized_at: i64,
    pub(crate) project_id: Option<String>,
    pub(crate) pre: Option<PreCheckpoint>,
    pub(crate) post: Option<PostCheckpoint>,
    pub(crate) pre_history: Vec<RetiredPreCheckpoint>,
    pub(crate) legacy_adoption_sha256: Option<String>,
    pub(crate) version: String,
    pub(crate) branch: String,
    pub(crate) base: String,
    pub(crate) base_commit: String,
    pub(crate) source_baseline: String,
    pub(crate) baseline: String,
    pub(crate) worktree_identity: String,
    pub(crate) source_state_sha256: String,
    pub(crate) source_archive_sha256: String,
    pub(crate) target_state_sha256: String,
    pub(crate) source_archive: String,
    pub(crate) successor_incarnation: String,
    pub(crate) succeeded_at: i64,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct PendingSuccessor {
    pub(crate) run: String,
    pub(crate) source_incarnation: String,
    pub(crate) target_incarnation: String,
    pub(crate) version: String,
    pub(crate) branch: String,
    pub(crate) base: String,
    pub(crate) base_commit: String,
    pub(crate) source_baseline: String,
    pub(crate) baseline: String,
    pub(crate) worktree_identity: String,
    pub(crate) source_state_sha256: String,
    pub(crate) source_archive_sha256: String,
    pub(crate) target_state_sha256: String,
    pub(crate) source_archive: String,
    pub(crate) created_at: i64,
    pub(crate) abort_nonce: String,
    source_authority: RunAuthoritySnapshot,
    terminal_reservation: String,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct RunAuthoritySnapshot {
    run: String,
    project_id: Option<String>,
    run_incarnation: String,
    orientation_epoch: u64,
    initialized_at: i64,
    pre: Option<PreCheckpoint>,
    post: Option<PostCheckpoint>,
    pre_history: Vec<RetiredPreCheckpoint>,
    legacy_adoption_sha256: Option<String>,
    predecessors: Vec<RetiredRunIncarnation>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct SuccessorSpec {
    pub(crate) run: String,
    pub(crate) source_incarnation: String,
    pub(crate) target_incarnation: String,
    pub(crate) version: String,
    pub(crate) branch: String,
    pub(crate) base: String,
    pub(crate) base_commit: String,
    pub(crate) source_baseline: String,
    pub(crate) baseline: String,
    pub(crate) worktree_identity: String,
    pub(crate) source_state_sha256: String,
    pub(crate) source_archive_sha256: String,
    pub(crate) target_state_sha256: String,
    pub(crate) source_archive: String,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) enum SuccessorNativeState {
    Current(RunAuthority),
    Pending(PendingSuccessor),
    Complete(RetiredRunIncarnation),
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct AbortedSuccessor {
    pub(crate) run: String,
    pub(crate) source_incarnation: String,
    pub(crate) target_incarnation: String,
    pub(crate) intent_sha256: String,
    pub(crate) failure_archive: String,
    pub(crate) aborted_at: i64,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct PreCheckpoint {
    pub(crate) epoch: u64,
    pub(crate) created_at: i64,
    pub(crate) run_revision: u64,
    pub(crate) manifest_sha256: String,
    pub(crate) pre_sha256: String,
    pub(crate) engineer_nonce: String,
    #[serde(default)]
    pub(crate) input_archive: Option<ArchivedInput>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct ArchivedInput {
    pub(crate) path: String,
    pub(crate) sha256: String,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct RetiredPreCheckpoint {
    pub(crate) pre: PreCheckpoint,
    pub(crate) post: Option<PostCheckpoint>,
    pub(crate) replaced_agent_id: String,
    pub(crate) replacement_agent_id: String,
    pub(crate) replacement_nonce: String,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct PostCheckpoint {
    pub(crate) epoch: u64,
    pub(crate) created_at: i64,
    pub(crate) run_revision: u64,
    pub(crate) post_sha256: String,
    pub(crate) critic_nonce: String,
    #[serde(default)]
    pub(crate) input_archive: Option<ArchivedInput>,
}

pub(crate) fn registry_path(context: &ExecutionContext) -> PathBuf {
    context.namespace.join(REGISTRY_FILE)
}

pub(crate) fn registry_lock_path(context: &ExecutionContext) -> PathBuf {
    context.namespace.join(LOCK_FILE)
}

pub(crate) fn new_incarnation() -> String {
    Uuid::now_v7().simple().to_string()
}

pub(crate) fn new_nonce() -> String {
    format!("{}{}", Uuid::now_v7().simple(), Uuid::now_v7().simple())
}

pub(crate) fn register_run(
    context: &ExecutionContext,
    run: &str,
    now: i64,
) -> Result<RunAuthority, CliError> {
    let project_root = canonical(&context.primary_root)?;
    let project_id = crate::cmd::dispatch::read_project_id(&context.project_id_path)
        .ok()
        .map(|value| value.to_string());
    with_registry(context, true, |registry| {
        if registry.project_root != project_root {
            return Err(error(
                "native orientation registry belongs to another project root",
            ));
        }
        if registry.runs.contains_key(run) {
            return Err(error(format!(
                "run incarnation for `{run}` already exists in native registry"
            )));
        }
        let authority = RunAuthority {
            run: run.into(),
            project_id,
            run_incarnation: new_incarnation(),
            orientation_epoch: 0,
            initialized_at: now,
            pre: None,
            post: None,
            pre_history: Vec::new(),
            legacy_adoption_sha256: None,
            predecessors: Vec::new(),
        };
        registry.runs.insert(run.into(), authority.clone());
        Ok(authority)
    })
}

/// Adopt only a pristine legacy run while the caller holds its Native run
/// lock. Publish an exact source-bound intent before its RunState incarnation.
/// If the latter write fails, ordinary authority checks remain fail-closed;
/// only this explicit command may resume the same unchanged source intent.
pub(crate) fn adopt_legacy_state(
    context: &ExecutionContext,
    run: &str,
    state: &mut RunState,
    access: &crate::run_store::RunAccess<'_>,
    now: i64,
) -> Result<String, CliError> {
    if state.run != run
        || !state.status.is(RunStatus::Planted)
        || state.orientation_epoch != 0
        || !state.lanes.is_empty()
    {
        return Err(error(
            "native adoption requires a pristine planted run with zero orientation epoch and no lanes",
        ));
    }
    let run_id = shepherd::dispatch::RunId::new(run).map_err(|source| error(source.to_string()))?;
    let inventory = crate::DispatchStore::new(&context.runs_root)
        .with_run_access(&run_id, access, |authority| authority.inventory())
        .map_err(|source| error(source.to_string()))?;
    if !inventory.records.is_empty() || !inventory.pending.is_empty() {
        return Err(error(
            "native adoption refuses existing child dispatch or pending authority",
        ));
    }
    let project_id = if context.project_id_path.exists() {
        Some(crate::cmd::dispatch::read_project_id(&context.project_id_path)?.to_string())
    } else {
        None
    };
    let mut legacy = state.clone();
    legacy.run_incarnation.clear();
    let source_sha256 = Sha256::digest(legacy.to_canonical_json().as_bytes())
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect::<String>();
    let incarnation = with_registry(context, true, |registry| {
        if registry.schema != REGISTRY_SCHEMA
            || registry.policy != POLICY
            || registry.project_root != canonical(&context.primary_root)?
        {
            return Err(error("native adoption registry identity or policy changed"));
        }
        if let Some(current) = registry.runs.get(run) {
            validate_registry_identity(
                &context.primary_root,
                registry,
                current,
                run,
                project_id.as_deref(),
            )?;
            if current.legacy_adoption_sha256.as_deref() != Some(source_sha256.as_str())
                || current.orientation_epoch != 0
                || current.pre.is_some()
                || current.post.is_some()
                || !current.pre_history.is_empty()
                || current.project_id != project_id
                || (!state.run_incarnation.is_empty()
                    && state.run_incarnation != current.run_incarnation)
            {
                return Err(error(
                    "native adoption intent does not match this pristine legacy source",
                ));
            }
            return Ok(current.run_incarnation.clone());
        }
        if !state.run_incarnation.is_empty() {
            return Err(error(
                "native adoption cannot import a caller-chosen run incarnation",
            ));
        }
        let incarnation = new_incarnation();
        registry.runs.insert(
            run.into(),
            RunAuthority {
                run: run.into(),
                project_id,
                run_incarnation: incarnation.clone(),
                orientation_epoch: 0,
                initialized_at: now,
                pre: None,
                post: None,
                pre_history: Vec::new(),
                legacy_adoption_sha256: Some(source_sha256),
                predecessors: Vec::new(),
            },
        );
        Ok(incarnation)
    })?;
    state.run_incarnation = incarnation.clone();
    Ok(incarnation)
}

/// Read the one Native successor state for an exact source incarnation.
/// This is the only read allowed while a durable successor intent is pending;
/// ordinary lifecycle readers fail closed in [`load_run_at`].
pub(crate) fn successor_state(
    context: &ExecutionContext,
    run: &str,
    source_incarnation: &str,
) -> Result<SuccessorNativeState, CliError> {
    let registry = read_registry(context)?;
    let current = registry.runs.get(run).cloned().ok_or_else(|| {
        error(format!(
            "native registry has no incarnation for run `{run}`"
        ))
    })?;
    validate_registry_identity(&context.primary_root, &registry, &current, run, None)?;
    if let Some(pending) = registry.pending_successors.get(run) {
        if pending.source_incarnation != source_incarnation {
            return Err(error(format!(
                "pending successor source incarnation is {}, observed {source_incarnation}",
                pending.source_incarnation
            )));
        }
        return Ok(SuccessorNativeState::Pending(pending.clone()));
    }
    if current.run_incarnation == source_incarnation {
        return Ok(SuccessorNativeState::Current(current));
    }
    let mut matches = current
        .predecessors
        .iter()
        .filter(|retired| retired.run_incarnation == source_incarnation);
    let Some(retired) = matches.next().cloned() else {
        return Err(error(format!(
            "source incarnation mismatch: expected {}, observed {source_incarnation}",
            current.run_incarnation
        )));
    };
    if matches.next().is_some() {
        return Err(error(
            "native successor history is ambiguous for the requested source incarnation",
        ));
    }
    Ok(SuccessorNativeState::Complete(retired))
}

pub(crate) fn ensure_no_pending_successor(
    context: &ExecutionContext,
    run: &str,
) -> Result<(), CliError> {
    let path = registry_path(context);
    if !path.exists() {
        return Ok(());
    }
    let registry = read_registry(context)?;
    if let Some(pending) = registry.pending_successors.get(run) {
        return Err(error(format!(
            "run `{run}` has an incomplete successor transition from incarnation {} to {}; rerun the exact `shepherd run successor` command before any mutation",
            pending.source_incarnation, pending.target_incarnation
        )));
    }
    Ok(())
}

/// Publish a durable successor intent before the caller moves any run
/// directory. An exact replay returns the same generated target incarnation;
/// a conflicting request fails without changing Native authority.
pub(crate) fn prepare_successor(
    context: &ExecutionContext,
    source: &RunAuthority,
    spec: &SuccessorSpec,
    now: i64,
) -> Result<PendingSuccessor, CliError> {
    with_registry(context, true, |registry| {
        let current = registry.runs.get(&spec.run).ok_or_else(|| {
            error(format!(
                "native registry has no incarnation for run `{}`",
                spec.run
            ))
        })?;
        if let Some(pending) = registry.pending_successors.get(&spec.run) {
            if pending_matches_spec(pending, spec) {
                return Ok(pending.clone());
            }
            return Err(error(
                "a different successor intent already exists for this canonical run",
            ));
        }
        if registry
            .aborted_successors
            .get(&spec.run)
            .map_or(0, Vec::len)
            >= MAX_ABORTED_SUCCESSORS_PER_RUN
        {
            return Err(error(format!(
                "run `{}` reached the bounded aborted-successor history limit of {MAX_ABORTED_SUCCESSORS_PER_RUN}",
                spec.run
            )));
        }
        if current != source
            || current.run_incarnation != spec.source_incarnation
            || current.run != spec.run
        {
            return Err(error(
                "native source authority changed before successor preparation",
            ));
        }
        let pending = PendingSuccessor {
            run: spec.run.clone(),
            source_incarnation: spec.source_incarnation.clone(),
            target_incarnation: spec.target_incarnation.clone(),
            version: spec.version.clone(),
            branch: spec.branch.clone(),
            base: spec.base.clone(),
            base_commit: spec.base_commit.clone(),
            source_baseline: spec.source_baseline.clone(),
            baseline: spec.baseline.clone(),
            worktree_identity: spec.worktree_identity.clone(),
            source_state_sha256: spec.source_state_sha256.clone(),
            source_archive_sha256: spec.source_archive_sha256.clone(),
            target_state_sha256: spec.target_state_sha256.clone(),
            source_archive: spec.source_archive.clone(),
            created_at: now,
            abort_nonce: new_nonce(),
            source_authority: authority_snapshot(source),
            terminal_reservation: "x".repeat(SUCCESSOR_TERMINAL_RESERVATION_BYTES),
        };
        registry
            .pending_successors
            .insert(spec.run.clone(), pending.clone());
        let prepared_len = registry_encoded_len(registry)?;
        let mut completed = registry.clone();
        apply_successor_completion(&mut completed, &pending, now)?;
        let completion_len = registry_encoded_len(&completed)?;
        let failure_archive = Path::new(&pending.source_archive)
            .parent()
            .ok_or_else(|| error("successor archive has no incarnation parent"))?;
        let mut aborted = registry.clone();
        apply_successor_abort(&mut aborted, &pending, failure_archive, now)?;
        let abort_len = registry_encoded_len(&aborted)?;
        if completion_len.max(abort_len).saturating_add(128) > prepared_len {
            return Err(error(format!(
                "pending successor does not reserve enough Native registry space for completion and abort: pending={prepared_len}, completion={completion_len}, abort={abort_len}"
            )));
        }
        Ok(pending)
    })
}

/// Atomically replace Native current authority after the caller proves that
/// the archived source and fresh active run match the prepared intent.
pub(crate) fn complete_successor(
    context: &ExecutionContext,
    prepared: &PendingSuccessor,
    now: i64,
) -> Result<RetiredRunIncarnation, CliError> {
    with_registry(context, true, |registry| {
        apply_successor_completion(registry, prepared, now)
    })
}

pub(crate) fn abort_successor(
    context: &ExecutionContext,
    prepared: &PendingSuccessor,
    failure_archive: &Path,
    now: i64,
) -> Result<AbortedSuccessor, CliError> {
    with_registry(context, true, |registry| {
        apply_successor_abort(registry, prepared, failure_archive, now)
    })
}

fn apply_successor_completion(
    registry: &mut Registry,
    prepared: &PendingSuccessor,
    now: i64,
) -> Result<RetiredRunIncarnation, CliError> {
    if prepared.terminal_reservation.len() != SUCCESSOR_TERMINAL_RESERVATION_BYTES {
        return Err(error("native successor terminal reservation changed"));
    }
    let pending = registry
        .pending_successors
        .get(&prepared.run)
        .ok_or_else(|| error("native successor intent disappeared before completion"))?;
    if pending != prepared {
        return Err(error("native successor intent changed before completion"));
    }
    let current = registry
        .runs
        .get(&prepared.run)
        .cloned()
        .ok_or_else(|| error("native source authority disappeared before completion"))?;
    if authority_snapshot(&current) != prepared.source_authority {
        return Err(error(
            "native source authority changed before successor completion",
        ));
    }
    let retired = RetiredRunIncarnation {
        run_incarnation: current.run_incarnation.clone(),
        orientation_epoch: current.orientation_epoch,
        initialized_at: current.initialized_at,
        project_id: current.project_id.clone(),
        pre: current.pre.clone(),
        post: current.post.clone(),
        pre_history: current.pre_history.clone(),
        legacy_adoption_sha256: current.legacy_adoption_sha256.clone(),
        version: prepared.version.clone(),
        branch: prepared.branch.clone(),
        base: prepared.base.clone(),
        base_commit: prepared.base_commit.clone(),
        source_baseline: prepared.source_baseline.clone(),
        baseline: prepared.baseline.clone(),
        worktree_identity: prepared.worktree_identity.clone(),
        source_state_sha256: prepared.source_state_sha256.clone(),
        source_archive_sha256: prepared.source_archive_sha256.clone(),
        target_state_sha256: prepared.target_state_sha256.clone(),
        source_archive: prepared.source_archive.clone(),
        successor_incarnation: prepared.target_incarnation.clone(),
        succeeded_at: now,
    };
    let mut predecessors = current.predecessors;
    predecessors.push(retired.clone());
    registry.runs.insert(
        prepared.run.clone(),
        RunAuthority {
            run: prepared.run.clone(),
            project_id: current.project_id,
            run_incarnation: prepared.target_incarnation.clone(),
            orientation_epoch: 0,
            initialized_at: now,
            pre: None,
            post: None,
            pre_history: Vec::new(),
            legacy_adoption_sha256: None,
            predecessors,
        },
    );
    registry.pending_successors.remove(&prepared.run);
    Ok(retired)
}

fn apply_successor_abort(
    registry: &mut Registry,
    prepared: &PendingSuccessor,
    failure_archive: &Path,
    now: i64,
) -> Result<AbortedSuccessor, CliError> {
    if prepared.terminal_reservation.len() != SUCCESSOR_TERMINAL_RESERVATION_BYTES {
        return Err(error("native successor terminal reservation changed"));
    }
    let pending = registry
        .pending_successors
        .get(&prepared.run)
        .ok_or_else(|| error("native successor intent is not pending"))?;
    if pending != prepared {
        return Err(error("native successor intent changed before abort"));
    }
    let current = registry
        .runs
        .get(&prepared.run)
        .ok_or_else(|| error("native source authority disappeared before abort"))?;
    if authority_snapshot(current) != prepared.source_authority {
        return Err(error(
            "native source authority changed before successor abort",
        ));
    }
    let intent_bytes = serde_json::to_vec(prepared)
        .map_err(|error| self::error(format!("encode successor abort intent: {error}")))?;
    let aborted = AbortedSuccessor {
        run: prepared.run.clone(),
        source_incarnation: prepared.source_incarnation.clone(),
        target_incarnation: prepared.target_incarnation.clone(),
        intent_sha256: Sha256::digest(intent_bytes)
            .iter()
            .map(|byte| format!("{byte:02x}"))
            .collect(),
        failure_archive: failure_archive.display().to_string(),
        aborted_at: now,
    };
    registry
        .aborted_successors
        .entry(prepared.run.clone())
        .or_default()
        .push(aborted.clone());
    registry.pending_successors.remove(&prepared.run);
    Ok(aborted)
}

fn registry_encoded_len(registry: &Registry) -> Result<usize, CliError> {
    serde_json::to_vec_pretty(registry)
        .map(|bytes| bytes.len().saturating_add(1))
        .map_err(|error| self::error(format!("encode Native successor registry: {error}")))
}

pub(crate) fn successor_matches_spec(
    retired: &RetiredRunIncarnation,
    spec: &SuccessorSpec,
) -> bool {
    retired.run_incarnation == spec.source_incarnation
        && retired.successor_incarnation == spec.target_incarnation
        && retired.version == spec.version
        && retired.branch == spec.branch
        && retired.base == spec.base
        && retired.base_commit == spec.base_commit
        && retired.source_baseline == spec.source_baseline
        && retired.baseline == spec.baseline
        && retired.worktree_identity == spec.worktree_identity
        && retired.source_state_sha256 == spec.source_state_sha256
        && retired.source_archive_sha256 == spec.source_archive_sha256
        && retired.target_state_sha256 == spec.target_state_sha256
        && retired.source_archive == spec.source_archive
}

pub(crate) fn pending_matches_spec(pending: &PendingSuccessor, spec: &SuccessorSpec) -> bool {
    pending.run == spec.run
        && pending.source_incarnation == spec.source_incarnation
        && pending.target_incarnation == spec.target_incarnation
        && pending.version == spec.version
        && pending.branch == spec.branch
        && pending.base == spec.base
        && pending.base_commit == spec.base_commit
        && pending.source_baseline == spec.source_baseline
        && pending.baseline == spec.baseline
        && pending.worktree_identity == spec.worktree_identity
        && pending.source_state_sha256 == spec.source_state_sha256
        && pending.source_archive_sha256 == spec.source_archive_sha256
        && pending.target_state_sha256 == spec.target_state_sha256
        && pending.source_archive == spec.source_archive
}

fn authority_snapshot(authority: &RunAuthority) -> RunAuthoritySnapshot {
    RunAuthoritySnapshot {
        run: authority.run.clone(),
        project_id: authority.project_id.clone(),
        run_incarnation: authority.run_incarnation.clone(),
        orientation_epoch: authority.orientation_epoch,
        initialized_at: authority.initialized_at,
        pre: authority.pre.clone(),
        post: authority.post.clone(),
        pre_history: authority.pre_history.clone(),
        legacy_adoption_sha256: authority.legacy_adoption_sha256.clone(),
        predecessors: authority.predecessors.clone(),
    }
}

pub(crate) fn remove_run(context: &ExecutionContext, run: &str) -> Result<(), CliError> {
    let path = registry_path(context);
    if !path.exists() {
        return Ok(());
    }
    with_registry(context, true, |registry| {
        if registry.pending_successors.contains_key(run) {
            return Err(error(
                "cannot remove a run while its successor transition is pending",
            ));
        }
        registry.runs.remove(run);
        Ok(())
    })
}

pub(crate) fn rename_run(
    context: &ExecutionContext,
    old: &str,
    new: &str,
) -> Result<bool, CliError> {
    let path = registry_path(context);
    if !path.exists() {
        return Ok(false);
    }
    with_registry(context, true, |registry| {
        if registry.pending_successors.contains_key(old) {
            return Err(error(
                "cannot rename a run while its successor transition is pending",
            ));
        }
        if registry.runs.contains_key(new) {
            return Err(error(format!(
                "native registry already contains an incarnation for `{new}`"
            )));
        }
        let Some(mut authority) = registry.runs.remove(old) else {
            return Ok(false);
        };
        authority.run = new.into();
        registry.runs.insert(new.into(), authority);
        Ok(true)
    })
}

pub(crate) fn load_run(
    context: &ExecutionContext,
    run: &str,
    project_id: Option<&str>,
) -> Result<RunAuthority, CliError> {
    load_run_at(&context.primary_root, run, project_id)
}

pub(crate) fn load_run_at(
    project_root: &Path,
    run: &str,
    project_id: Option<&str>,
) -> Result<RunAuthority, CliError> {
    let registry_path = project_root.join(".shepherd").join(REGISTRY_FILE);
    let registry = read_registry_path(&registry_path)?;
    if let Some(pending) = registry.pending_successors.get(run) {
        return Err(error(format!(
            "run `{run}` has an incomplete successor transition from incarnation {} to {}; rerun the exact `shepherd run successor` command",
            pending.source_incarnation, pending.target_incarnation
        )));
    }
    let authority = registry.runs.get(run).cloned().ok_or_else(|| {
        error(format!(
            "native registry has no incarnation for run `{run}`"
        ))
    })?;
    validate_registry_identity(project_root, &registry, &authority, run, project_id)?;
    Ok(authority)
}

pub(crate) fn validate_state(
    context: &ExecutionContext,
    run: &str,
    state: &RunState,
    project_id: Option<&str>,
) -> Result<RunAuthority, CliError> {
    let authority = load_run(context, run, project_id)?;
    if state.run != run || state.run_incarnation != authority.run_incarnation {
        return Err(error("run state does not match the native run incarnation"));
    }
    if state.orientation_epoch != authority.orientation_epoch {
        return Err(error(
            "run state orientation epoch does not match native registry",
        ));
    }
    Ok(authority)
}

pub(crate) fn validate_and_bind_state(
    context: &ExecutionContext,
    run: &str,
    state: &RunState,
    project_id: &str,
) -> Result<RunAuthority, CliError> {
    let authority = validate_state(context, run, state, None)?;
    if authority.project_id.is_none() {
        bind_project(context, run, project_id)
    } else {
        validate_state(context, run, state, Some(project_id))
    }
}

pub(crate) fn validate_and_bind_pre_state(
    context: &ExecutionContext,
    run: &str,
    state: &RunState,
    project_id: &str,
) -> Result<RunAuthority, CliError> {
    let mut authority = load_run(context, run, None)?;
    if state.run != run || state.run_incarnation != authority.run_incarnation {
        return Err(error("run state does not match the native run incarnation"));
    }
    if authority.project_id.is_none() {
        if state.orientation_epoch != authority.orientation_epoch {
            return Err(error(
                "unbound run state orientation epoch does not match native registry",
            ));
        }
        authority = bind_project(context, run, project_id)?;
    } else if authority.project_id.as_deref() != Some(project_id) {
        return Err(error(
            "native run authority belongs to another project identity",
        ));
    }
    let recoverable_pre = authority.pre.as_ref().is_some_and(|checkpoint| {
        state.orientation_epoch.checked_add(1) == Some(authority.orientation_epoch)
            && checkpoint.epoch == authority.orientation_epoch
            && checkpoint.run_revision == authority.orientation_epoch
    });
    if state.orientation_epoch != authority.orientation_epoch && !recoverable_pre {
        return Err(error(
            "run state orientation epoch does not match native registry",
        ));
    }
    Ok(authority)
}

pub(crate) fn bind_project(
    context: &ExecutionContext,
    run: &str,
    project_id: &str,
) -> Result<RunAuthority, CliError> {
    with_registry(context, true, |registry| {
        let authority = registry.runs.get_mut(run).ok_or_else(|| {
            error(format!(
                "native registry has no incarnation for run `{run}`"
            ))
        })?;
        if authority.project_id.is_some() && authority.project_id.as_deref() != Some(project_id) {
            return Err(error(
                "native run authority belongs to another project identity",
            ));
        }
        authority.project_id = Some(project_id.into());
        Ok(authority.clone())
    })
}

pub(crate) fn record_pre(
    context: &ExecutionContext,
    run: &str,
    authority: &RunAuthority,
    checkpoint: PreCheckpoint,
) -> Result<RunAuthority, CliError> {
    with_registry(context, true, |registry| {
        let current = registry.runs.get_mut(run).ok_or_else(|| {
            error(format!(
                "native registry has no incarnation for run `{run}`"
            ))
        })?;
        if current != authority || current.pre.is_some() {
            return Err(error("native pre checkpoint authority changed"));
        }
        if checkpoint.epoch != current.orientation_epoch + 1 {
            return Err(error("native orientation epoch is not monotonic"));
        }
        current.orientation_epoch = checkpoint.epoch;
        current.pre = Some(checkpoint);
        current.post = None;
        Ok(current.clone())
    })
}

pub(crate) fn record_post(
    context: &ExecutionContext,
    run: &str,
    authority: &RunAuthority,
    checkpoint: PostCheckpoint,
) -> Result<RunAuthority, CliError> {
    with_registry(context, true, |registry| {
        let current = registry.runs.get_mut(run).ok_or_else(|| {
            error(format!(
                "native registry has no incarnation for run `{run}`"
            ))
        })?;
        if current != authority || current.pre.is_none() {
            return Err(error("native post checkpoint has no matching native pre"));
        }
        if checkpoint.epoch != current.orientation_epoch {
            return Err(error("native post checkpoint epoch is stale"));
        }
        current.post = Some(checkpoint);
        Ok(current.clone())
    })
}

/// The orientation caller has checked exact Native replacement lineage while
/// holding the selected run lock. Retain both checkpoints before a new epoch.
pub(crate) fn retire_pre(
    context: &ExecutionContext,
    run: &str,
    authority: &RunAuthority,
    retired: RetiredPreCheckpoint,
) -> Result<RunAuthority, CliError> {
    with_registry(context, true, |registry| {
        let current = registry
            .runs
            .get_mut(run)
            .ok_or_else(|| error("native run is absent"))?;
        if current != authority
            || current.pre.as_ref() != Some(&retired.pre)
            || current.post != retired.post
            || retired.pre.epoch != current.orientation_epoch
            || retired.pre.input_archive.is_none()
            || retired.replaced_agent_id.is_empty()
            || retired.replacement_agent_id.is_empty()
            || retired.replaced_agent_id == retired.replacement_agent_id
            || retired.replacement_nonce.is_empty()
            || retired.replacement_nonce == retired.pre.engineer_nonce
        {
            return Err(error(
                "native pre retirement does not match exact replacement custody",
            ));
        }
        current.pre_history.push(retired);
        current.pre = None;
        current.post = None;
        Ok(current.clone())
    })
}

fn validate_registry_identity(
    project_root: &Path,
    registry: &Registry,
    authority: &RunAuthority,
    run: &str,
    project_id: Option<&str>,
) -> Result<(), CliError> {
    if registry.schema != REGISTRY_SCHEMA || registry.policy != POLICY || authority.run != run {
        return Err(error(
            "native orientation registry has the wrong schema or policy",
        ));
    }
    if registry.project_root != canonical(project_root)? {
        return Err(error("native orientation registry project root changed"));
    }
    if let (Some(expected), Some(bound)) = (project_id, authority.project_id.as_deref())
        && bound != expected
    {
        return Err(error("native run authority project identity changed"));
    }
    if authority.run_incarnation.is_empty() {
        return Err(error("native run authority has no incarnation"));
    }
    Ok(())
}

fn read_registry(context: &ExecutionContext) -> Result<Registry, CliError> {
    read_registry_path(&registry_path(context))
}

fn read_registry_path(path: &Path) -> Result<Registry, CliError> {
    let bytes = read_nofollow(path)?;
    serde_json::from_slice(&bytes)
        .map_err(|source| error(format!("native orientation registry is invalid: {source}")))
}

fn with_registry<T>(
    context: &ExecutionContext,
    create: bool,
    mutate: impl FnOnce(&mut Registry) -> Result<T, CliError>,
) -> Result<T, CliError> {
    let path = registry_path(context);
    let parent = path
        .parent()
        .ok_or_else(|| error("native orientation registry has no parent"))?;
    if create {
        fs::create_dir_all(parent)
            .map_err(|source| error(format!("create registry parent: {source}")))?;
    }
    reject_path(parent)?;
    let lock_path = registry_lock_path(context);
    let lock = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .truncate(false)
        .open(&lock_path)
        .map_err(|source| error(format!("open native registry lock: {source}")))?;
    acquire_lock(&lock, &lock_path)?;
    let mut registry = if path.exists() {
        read_registry(context)?
    } else {
        Registry {
            schema: REGISTRY_SCHEMA.into(),
            policy: POLICY.into(),
            project_root: canonical(&context.primary_root)?,
            runs: BTreeMap::new(),
            pending_successors: BTreeMap::new(),
            aborted_successors: BTreeMap::new(),
        }
    };
    let result = mutate(&mut registry)?;
    write_atomic(&path, &registry)?;
    Ok(result)
}

fn acquire_lock(file: &File, path: &Path) -> Result<(), CliError> {
    let started = std::time::Instant::now();
    loop {
        match file.try_lock() {
            Ok(()) => return Ok(()),
            Err(std::fs::TryLockError::WouldBlock) if started.elapsed() < LOCK_TIMEOUT => {
                std::thread::sleep(Duration::from_millis(10));
            }
            Err(std::fs::TryLockError::WouldBlock) => {
                return Err(error(format!(
                    "timed out waiting for native registry lock {}",
                    path.display()
                )));
            }
            Err(std::fs::TryLockError::Error(source)) => {
                return Err(error(format!("acquire native registry lock: {source}")));
            }
        }
    }
}

fn write_atomic<T: Serialize>(path: &Path, value: &T) -> Result<(), CliError> {
    reject_path(path)?;
    let parent = path
        .parent()
        .ok_or_else(|| error("registry has no parent"))?;
    let mut bytes = serde_json::to_vec_pretty(value).map_err(|source| error(source.to_string()))?;
    bytes.push(b'\n');
    if u64::try_from(bytes.len()).map_or(true, |length| length > MAX_BYTES) {
        return Err(error(
            "native orientation registry exceeds its bounded reader",
        ));
    }
    let name = path
        .file_name()
        .and_then(|value| value.to_str())
        .ok_or_else(|| error("registry filename is not UTF-8"))?;
    let temporary = parent.join(format!(".{name}.tmp-{}", std::process::id()));
    if temporary.exists() {
        return Err(error("native registry temporary path already exists"));
    }
    let mut file = OpenOptions::new()
        .write(true)
        .create_new(true)
        .open(&temporary)
        .map_err(|source| error(format!("create native registry temporary: {source}")))?;
    file.write_all(&bytes)
        .and_then(|_| file.sync_all())
        .map_err(|source| error(format!("write native registry: {source}")))?;
    fs::rename(&temporary, path).map_err(|source| {
        let _ = fs::remove_file(&temporary);
        error(format!("publish native registry: {source}"))
    })?;
    // Fsyncing the parent directory is what makes the rename durable on POSIX.
    // Windows has no equivalent: opening a directory handle needs backup
    // semantics, and NTFS refuses FlushFileBuffers on one either way, so both
    // halves fail with "Access is denied" (os error 5). The rename is already
    // ordered by the NTFS log, so the non-unix twin has nothing to do.
    #[cfg(unix)]
    fs::File::open(parent)
        .and_then(|directory| directory.sync_all())
        .map_err(|source| error(format!("sync native registry parent: {source}")))?;
    Ok(())
}

fn read_nofollow(path: &Path) -> Result<Vec<u8>, CliError> {
    reject_path(path)?;
    let mut file = OpenOptions::new()
        .read(true)
        .open(path)
        .map_err(|source| error(format!("open native orientation registry: {source}")))?;
    let before = file
        .metadata()
        .map_err(|source| error(format!("inspect native orientation registry: {source}")))?;
    if !before.is_file() || before.len() > MAX_BYTES {
        return Err(error(
            "native orientation registry is not a bounded regular file",
        ));
    }
    let mut bytes = Vec::new();
    (&mut file)
        .take(MAX_BYTES + 1)
        .read_to_end(&mut bytes)
        .map_err(|source| error(format!("read native orientation registry: {source}")))?;
    if bytes.len() as u64 > MAX_BYTES {
        return Err(error("native orientation registry exceeds its byte limit"));
    }
    let after = file
        .metadata()
        .map_err(|source| error(format!("inspect native orientation registry: {source}")))?;
    if before.len() != after.len() || before.modified().ok() != after.modified().ok() {
        return Err(error(
            "native orientation registry changed while it was read",
        ));
    }
    Ok(bytes)
}

fn canonical(path: &Path) -> Result<String, CliError> {
    reject_path(path)?;
    fs::canonicalize(path)
        .map(|value| value.display().to_string())
        .map_err(|source| error(format!("canonicalize {}: {source}", path.display())))
}

fn reject_path(path: &Path) -> Result<(), CliError> {
    use std::path::Component;
    let mut cursor = PathBuf::new();
    for component in path.components() {
        cursor.push(component.as_os_str());
        if matches!(component, Component::Prefix(_) | Component::RootDir) {
            continue;
        }
        match fs::symlink_metadata(&cursor) {
            Ok(metadata) if metadata.file_type().is_symlink() => {
                return Err(error(format!(
                    "native authority path is a symlink: {}",
                    cursor.display()
                )));
            }
            Ok(_) => {}
            Err(source) if source.kind() == std::io::ErrorKind::NotFound => break,
            Err(source) => return Err(error(format!("inspect {}: {source}", cursor.display()))),
        }
    }
    Ok(())
}

fn error(message: impl Into<String>) -> CliError {
    CliError::message(message.into())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn interrupted_adoption_reuses_the_committed_intent_without_granting_early_authority() {
        let fixture = tempfile::tempdir().unwrap();
        let root = fs::canonicalize(fixture.path()).unwrap();
        assert!(
            std::process::Command::new("git")
                .args(["init", "--quiet"])
                .current_dir(&root)
                .status()
                .unwrap()
                .success()
        );
        let run_dir = root.join(".shepherd/runs/v657");
        fs::create_dir_all(run_dir.join("dispatch")).unwrap();
        let context = ExecutionContext::discover(crate::ContextInputs {
            start_dir: root.clone(),
            shepherd_home: Some(root.join("isolated-home")),
            home_dir: Some(root.join("isolated-home")),
            ..Default::default()
        })
        .unwrap();
        let state: RunState = serde_json::from_value(serde_json::json!({
            "run":"v657", "status":"planted", "updated_at":42,
        }))
        .unwrap();
        let store = crate::RunStore::new(run_dir.join("run.json"));
        store.initialize(&state).unwrap();
        let before = fs::read(store.path()).unwrap();
        let interrupted: crate::run_store::RunStoreResult<()> =
            store.update_with_access(|state, access| {
                adopt_legacy_state(&context, "v657", state, access, 100).unwrap();
                Err(crate::RunStoreError::mutation(
                    "injected interruption before RunState publication",
                ))
            });
        assert!(interrupted.is_err());
        assert_eq!(before, fs::read(store.path()).unwrap());
        let intent = load_run(&context, "v657", None).unwrap();
        assert!(validate_state(&context, "v657", &store.load().unwrap(), None).is_err());
        let recovered = store
            .update_with_access(|state, access| {
                Ok(adopt_legacy_state(&context, "v657", state, access, 200).unwrap())
            })
            .unwrap();
        assert_eq!(recovered, intent.run_incarnation);
        assert_eq!(load_run(&context, "v657", None).unwrap(), intent);
        assert!(validate_state(&context, "v657", &store.load().unwrap(), None).is_ok());
        assert_eq!(store.load().unwrap().updated_at, 42);
    }

    #[test]
    fn native_tokens_are_nonempty_and_distinct() {
        let incarnation = new_incarnation();
        let nonce = new_nonce();
        assert!(!incarnation.is_empty());
        assert!(!nonce.is_empty());
        assert_ne!(incarnation, nonce);
    }
}