shepherd-cli 6.6.1

//! Primary-run dispatch persistence with descriptor-relative path confinement.

use std::path::{Path, PathBuf};
use std::time::Duration;

use std::time::Instant;

#[cfg(unix)]
use std::{
    fs::File,
    io::{Read, Write},
    sync::atomic::{AtomicU64, Ordering},
    time::{SystemTime, UNIX_EPOCH},
};

use shepherd::dispatch::{
    AgentId, DispatchError, DispatchRecord, DispatchState, IdentityError, IdentityResolution,
    LaunchCleanupResponse, NativeIdentity, PendingDispatch, PendingLaunchState, ProfileLease,
    ReviewCustody, ReviewCustodyState, RootSessionBinding, RunId, SessionId, SkillUseChallenge,
    SkillUseRootAuthority, SkillUseState, StopRequest, constant_time_digest_eq,
    resolve_native_identity,
};

use shepherd::run::{RunStatus, Vocabulary};
use shepherd::{
    RunState,
    registry::{DispatchSingletonPublication, Registry, SingletonPublicationState},
};

pub type DispatchStoreResult<T> = core::result::Result<T, DispatchStoreError>;

#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum DispatchStoreError {
    #[error("dispatch filesystem operation `{operation}` failed for {}: {source}", path.display())]
    Io {
        operation: &'static str,
        path: PathBuf,
        #[source]
        source: std::io::Error,
    },
    #[error("unsafe dispatch path: {}", path.display())]
    UnsafePath { path: PathBuf },
    #[error("invalid run document {}: {reason}", path.display())]
    InvalidRunDocument { path: PathBuf, reason: String },
    #[error("no executing shepherd run exists")]
    NoActiveRun,
    #[error("multiple executing shepherd runs are ambiguous: {runs:?}")]
    AmbiguousActiveRuns { runs: Vec<RunId> },
    #[error("dispatch record already exists: {}", path.display())]
    AlreadyExists { path: PathBuf },
    #[error("dispatch record is {size} bytes; maximum is {max} bytes")]
    RecordTooLarge { size: usize, max: usize },
    #[error("pending launch identity is not present for run `{run}`")]
    PendingNotFound { run: RunId },
    #[error("pending launch identity path is invalid for run `{run}`")]
    PendingPath { run: RunId },
    #[error("dispatch record for `{agent_id}` is unknown in run `{run}`: {reason}")]
    UnknownRecord {
        run: RunId,
        agent_id: AgentId,
        reason: String,
    },
    #[error("event names run `{supplied}`, but primary active run is `{active}`")]
    WrongActiveRun { supplied: RunId, active: RunId },
    #[error("timed out after {timeout:?} waiting for dispatch lock {}", path.display())]
    LockTimeout { path: PathBuf, timeout: Duration },
    #[error(transparent)]
    Domain(#[from] DispatchError),
    #[error(transparent)]
    Identity(#[from] IdentityError),
    #[error("singleton publication {nonce} failed: {reason}")]
    SingletonPublication { nonce: String, reason: String },
    #[error("singleton publication reconciliation failed: {0}")]
    Reconciliation(String),
}

impl DispatchStoreError {
    fn io(operation: &'static str, path: PathBuf, source: impl Into<std::io::Error>) -> Self {
        Self::Io {
            operation,
            path,
            source: source.into(),
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PendingClaim {
    pub pending: PendingDispatch,
}

/// One locked observation, without the normal loader's quarantine recovery.
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct ReviewTerminalSnapshot {
    pub(crate) root: RootSessionBinding,
    pub(crate) subject: DispatchRecord,
    pub(crate) pending: PendingDispatch,
    pub(crate) custody: ReviewCustody,
}

/// Native records observed together under the selected run's dispatch lock.
#[derive(Clone, Debug, Default)]
pub(crate) struct DispatchInventory {
    pub(crate) records: Vec<DispatchRecord>,
    pub(crate) pending: Vec<PendingDispatch>,
}

/// Authority reads made inside a pending transition's existing run lock.
/// The store constructs this borrowed view only while holding that lock; it
/// cannot be cloned or retained by the validation callback.
pub struct LockedDispatchRun<'a> {
    store: &'a DispatchStore,
    run: &'a RunId,
    #[cfg(unix)]
    dispatch: &'a rustix::fd::OwnedFd,
    #[cfg(windows)]
    dispatch: &'a Path,
}

impl LockedDispatchRun<'_> {
    pub(crate) fn inventory(&self) -> DispatchStoreResult<DispatchInventory> {
        platform::read_locked_inventory(self)
    }

    pub fn load_root_binding(
        &self,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        platform::read_locked_root_binding(self, session_id)
    }

    pub fn load_review_custody(&self, subject: &AgentId) -> DispatchStoreResult<ReviewCustody> {
        platform::read_locked_review_custody(self, subject)
    }

    pub(crate) fn review_terminal_snapshot(
        &self,
        subject: &AgentId,
    ) -> DispatchStoreResult<ReviewTerminalSnapshot> {
        let custody = self.load_review_custody(subject)?;
        Ok(ReviewTerminalSnapshot {
            root: self.load_root_binding(&custody.root_session_id)?,
            subject: platform::read_locked_record(self, subject)?,
            pending: platform::read_locked_pending(self, custody.pending_launch_id_hash)?,
            custody,
        })
    }
}

/// Deterministic failure points for the SQLite/filesystem publication seam.
#[derive(
    Clone,
    Copy,
    Debug,
    Eq,
    Hash,
    Ord,
    PartialEq,
    PartialOrd,
    strum::AsRefStr,
    strum::Display,
    strum::EnumCount,
    strum::EnumIs,
    strum::EnumString,
    strum::IntoStaticStr,
    strum::VariantNames,
)]
#[strum(ascii_case_insensitive, serialize_all = "snake_case")]
pub enum PublicationFault {
    AfterPreparing,
    AfterFilesystemPublish,
}

/// Deterministic crash seams for the malignant-custody commit protocol.
#[derive(
    Clone,
    Copy,
    Debug,
    Eq,
    Hash,
    Ord,
    PartialEq,
    PartialOrd,
    strum::AsRefStr,
    strum::Display,
    strum::EnumCount,
    strum::EnumIs,
    strum::EnumString,
    strum::IntoStaticStr,
    strum::VariantNames,
)]
#[strum(ascii_case_insensitive, serialize_all = "snake_case")]
pub enum ReviewQuarantineFault {
    AfterCustodyCommit,
    AfterRecordCommit,
}

/// Crash seams in a journaled same-principal root lease renewal.
#[derive(
    Clone,
    Copy,
    Debug,
    Eq,
    Hash,
    Ord,
    PartialEq,
    PartialOrd,
    strum::AsRefStr,
    strum::Display,
    strum::EnumCount,
    strum::EnumIs,
    strum::EnumString,
    strum::IntoStaticStr,
    strum::VariantNames,
)]
#[strum(ascii_case_insensitive, serialize_all = "snake_case")]
pub enum RootRenewalFault {
    AfterIntentPublish,
    AfterIntentCommit,
    AfterRunBindingCommit,
    AfterIndexCommit,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct RootRenewalIntent {
    schema: String,
    pub(crate) expected: RootSessionBinding,
    pub(crate) replacement: RootSessionBinding,
    pub(crate) profile_lease: Option<ProfileLease>,
}

impl RootRenewalIntent {
    fn validate(&self) -> DispatchStoreResult<()> {
        self.expected.validate()?;
        self.replacement.validate()?;
        let mut identity = self.expected.clone();
        identity.bound_at = self.replacement.bound_at;
        identity.expires_at = self.replacement.expires_at;
        if self.schema != "shepherd.root-renewal/1"
            || identity != self.replacement
            || self.replacement.bound_at <= self.expected.bound_at
            || self.replacement.expires_at <= self.expected.expires_at
            || self.replacement.expires_at - self.replacement.bound_at > 86_400_000
        {
            return Err(root_renewal_error(
                "root renewal is not a bounded same-identity lease extension",
            ));
        }
        if let Some(profile) = &self.profile_lease {
            profile.validate()?;
            if profile.project_id != self.expected.project_id
                || profile.run != self.expected.run
                || profile.root_session_id != self.expected.session_id
                || profile.harness != self.expected.harness
                || self.expected.project_filesystem_id.as_deref()
                    != Some(profile.project_filesystem_id.as_str())
            {
                return Err(root_renewal_error(
                    "root renewal profile belongs to a different identity",
                ));
            }
        }
        Ok(())
    }
}

fn root_renewal_error(message: &str) -> DispatchStoreError {
    DispatchStoreError::Reconciliation(message.into())
}

fn decode_root_renewal(
    bytes: &[u8],
    session: &SessionId,
) -> DispatchStoreResult<RootRenewalIntent> {
    let value: RootRenewalIntent = serde_json::from_slice(bytes)
        .map_err(|error| root_renewal_error(&format!("invalid root renewal intent: {error}")))?;
    value.validate()?;
    if &value.expected.session_id != session {
        return Err(root_renewal_error(
            "root renewal intent does not match its session path",
        ));
    }
    Ok(value)
}

/// Only old/old, new/old and new/new are reachable in this commit order.
/// Neither a new profile nor a different binding is recovery authority.
fn validate_root_renewal_observation(
    intent: &RootRenewalIntent,
    binding: &RootSessionBinding,
    index: &RootSessionBinding,
    profile: Option<&ProfileLease>,
    status: &Vocabulary<RunStatus>,
) -> DispatchStoreResult<(bool, bool)> {
    intent.validate()?;
    let run_old = binding == &intent.expected;
    let index_old = index == &intent.expected;
    if profile != intent.profile_lease.as_ref()
        || matches!(status.known(), Some(RunStatus::Closed | RunStatus::Closing))
        || !(run_old || binding == &intent.replacement)
        || !(index_old || index == &intent.replacement)
        || (run_old && !index_old)
    {
        return Err(root_renewal_error(
            "root renewal cannot reconcile changed identity, profile, or impossible commit order",
        ));
    }
    Ok((run_old, index_old))
}

/// Counts emitted by one replay pass over durable publication intents.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct ReconciliationReport {
    pub published: usize,
    pub refreshed: usize,
    pub quarantined: usize,
    pub unchanged: usize,
}

enum PublicationReconcile {
    Published,
    Refreshed,
    Quarantined(String),
    Unchanged,
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct DispatchStore {
    runs_root: PathBuf,
    timeout: Duration,
    publication_fault: Option<PublicationFault>,
    review_quarantine_fault: Option<ReviewQuarantineFault>,
    root_renewal_fault: Option<RootRenewalFault>,
}

impl DispatchStore {
    pub const DEFAULT_LOCK_TIMEOUT: Duration = Duration::from_secs(5);

    pub fn new(primary_runs_root: impl AsRef<Path>) -> Self {
        Self::with_timeout(primary_runs_root, Self::DEFAULT_LOCK_TIMEOUT)
    }

    pub fn with_timeout(primary_runs_root: impl AsRef<Path>, timeout: Duration) -> Self {
        Self {
            runs_root: primary_runs_root.as_ref().to_path_buf(),
            timeout,
            publication_fault: None,
            review_quarantine_fault: None,
            root_renewal_fault: None,
        }
    }

    /// Inject one deterministic boundary failure in tests. Production stores
    /// never carry a fault point.
    pub fn with_publication_fault(mut self, fault: PublicationFault) -> Self {
        self.publication_fault = Some(fault);
        self
    }

    /// Inject one deterministic review-quarantine crash boundary in tests.
    /// Production stores never carry a fault point.
    pub fn with_review_quarantine_fault(mut self, fault: ReviewQuarantineFault) -> Self {
        self.review_quarantine_fault = Some(fault);
        self
    }

    /// Inject a deterministic crash in tests; no CLI or hook accepts this input.
    pub fn with_root_renewal_fault(mut self, fault: RootRenewalFault) -> Self {
        self.root_renewal_fault = Some(fault);
        self
    }

    fn root_renewal_fault(&self, fault: RootRenewalFault) -> DispatchStoreResult<()> {
        if self.root_renewal_fault == Some(fault) {
            Err(root_renewal_error(&format!(
                "injected root renewal crash at {fault:?}"
            )))
        } else {
            Ok(())
        }
    }

    #[must_use]
    pub fn runs_root(&self) -> &Path {
        &self.runs_root
    }

    pub fn resolve_active_run(&self) -> DispatchStoreResult<RunId> {
        resolve_active_run(self)
    }

    pub fn load_run(&self, run: &RunId) -> DispatchStoreResult<RunState> {
        platform::load_run(self, run)
    }

    /// Load one explicitly selected run without turning absence into an
    /// infrastructure failure. Unsafe paths and malformed documents remain
    /// errors so callers can distinguish a missing run from corrupt authority.
    pub fn load_run_if_present(&self, run: &RunId) -> DispatchStoreResult<Option<RunState>> {
        match platform::load_run(self, run) {
            Ok(state) => Ok(Some(state)),
            Err(error) if is_not_found(&error) => Ok(None),
            Err(error) => Err(error),
        }
    }

    pub fn load_for_run(
        &self,
        run: &RunId,
        agent_id: &AgentId,
    ) -> DispatchStoreResult<DispatchRecord> {
        platform::load(self, run, agent_id)
    }

    pub fn load_root_binding_for_run(
        &self,
        run: &RunId,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        platform::load_root_binding(self, run, session_id)
    }

    /// Load the single current run binding for a root session. This index is
    /// outside every run directory so corrupt or linked sibling runs cannot
    /// influence explicit continuation authority.
    pub fn load_current_root_binding(
        &self,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        platform::load_current_root_binding(self, session_id)
    }

    /// Atomically move one trusted root session's authority to this binding.
    /// Older per-run records remain immutable evidence, but no operation may
    /// use them after this current-session index advances.
    pub fn activate_current_root_binding(
        &self,
        binding: &RootSessionBinding,
    ) -> DispatchStoreResult<()> {
        binding.validate()?;
        platform::activate_current_root_binding(self, binding)
    }

    /// Extend only an exact current root, preserving its complete identity.
    /// A durable intent gates readers until both binding records are committed.
    pub fn renew_root_binding(
        &self,
        expected: &RootSessionBinding,
        replacement: &RootSessionBinding,
        profile_lease: Option<&ProfileLease>,
    ) -> DispatchStoreResult<RootSessionBinding> {
        let intent = RootRenewalIntent {
            schema: "shepherd.root-renewal/1".into(),
            expected: expected.clone(),
            replacement: replacement.clone(),
            profile_lease: profile_lease.cloned(),
        };
        intent.validate()?;
        platform::commit_root_renewal(self, &intent, false)
    }

    pub(crate) fn root_renewal_intent(
        &self,
        session: &SessionId,
    ) -> DispatchStoreResult<Option<RootRenewalIntent>> {
        platform::root_renewal_intent(self, session)
    }

    /// Only the explicit, authenticated bind-root path invokes recovery.
    pub(crate) fn reconcile_root_renewal(
        &self,
        intent: &RootRenewalIntent,
    ) -> DispatchStoreResult<RootSessionBinding> {
        intent.validate()?;
        platform::commit_root_renewal(self, intent, true)
    }

    /// Resolve the newest valid explicit binding for one root session without
    /// consulting any run document. A root binding names its run directly;
    /// parsing unrelated sibling `run.json` files here would let stale or
    /// corrupt run state override that durable authority.
    pub fn load_latest_root_binding(
        &self,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        let root = platform::open_runs_root(self)?;
        let mut runs = platform::run_names(self, &root)?
            .into_iter()
            .filter_map(|name| RunId::new(name).ok())
            .collect::<Vec<_>>();
        runs.sort();
        runs.dedup();

        let mut latest: Option<RootSessionBinding> = None;
        let mut invalid = None;
        for run in runs {
            match platform::load_root_binding(self, &run, session_id) {
                Ok(binding) => {
                    let replace = latest.as_ref().is_none_or(|current| {
                        (binding.bound_at, binding.run.as_str())
                            > (current.bound_at, current.run.as_str())
                    });
                    if replace {
                        latest = Some(binding);
                    }
                }
                Err(error) if is_not_found(&error) => {}
                Err(error @ DispatchStoreError::Domain(DispatchError::InvalidRecord(_)))
                | Err(error @ DispatchStoreError::Identity(IdentityError::InvalidRootBinding(_))) =>
                {
                    invalid.get_or_insert(error);
                }
                Err(error) => return Err(error),
            }
        }
        latest
            .map(Ok)
            .or_else(|| invalid.map(Err))
            .unwrap_or_else(|| Err(IdentityError::MissingRootBinding.into()))
    }

    pub fn publish_pending(&self, pending: &PendingDispatch) -> DispatchStoreResult<()> {
        pending.validate()?;
        let state = self.load_run(&pending.run)?;
        if matches!(
            state.status.known(),
            Some(RunStatus::Closed | RunStatus::Closing)
        ) {
            return Err(DispatchStoreError::PendingPath {
                run: pending.run.clone(),
            });
        }
        platform::publish_pending(self, pending, None, |_, _, _| Ok(())).map(|_| ())
    }

    pub fn publish_pending_with_lease(
        &self,
        pending: &PendingDispatch,
        lease_ms: u64,
    ) -> DispatchStoreResult<PendingDispatch> {
        pending.validate()?;
        let state = self.load_run(&pending.run)?;
        if matches!(
            state.status.known(),
            Some(RunStatus::Closed | RunStatus::Closing)
        ) {
            return Err(DispatchStoreError::PendingPath {
                run: pending.run.clone(),
            });
        }
        platform::publish_pending(self, pending, Some(lease_ms), |_, _, _| Ok(()))
    }

    /// Check native capacity while holding the same lock that publishes the
    /// reservation. Two preparers cannot both spend the final process slot.
    pub(crate) fn publish_pending_bounded<F>(
        &self,
        pending: &PendingDispatch,
        lease_ms: u64,
        authorize: F,
    ) -> DispatchStoreResult<PendingDispatch>
    where
        F: FnOnce(&PendingDispatch, i64, &LockedDispatchRun<'_>) -> DispatchStoreResult<()>,
    {
        pending.validate()?;
        platform::publish_pending(self, pending, Some(lease_ms), authorize)
    }

    pub fn load_pending(
        &self,
        run: &RunId,
        launch_id_hash: [u8; 32],
    ) -> DispatchStoreResult<PendingDispatch> {
        platform::load_pending(self, run, launch_id_hash)
    }

    pub fn load_pending_for_agent(
        &self,
        run: &RunId,
        agent_id: &AgentId,
    ) -> DispatchStoreResult<PendingDispatch> {
        platform::load_pending_for_agent(self, run, agent_id)
    }

    pub fn claim_pending_unspawned<F>(
        &self,
        run: &RunId,
        launch_hash: [u8; 32],
        child_process_hash: [u8; 32],
        validate: F,
    ) -> DispatchStoreResult<PendingDispatch>
    where
        F: FnOnce(&PendingDispatch, i64, &LockedDispatchRun<'_>) -> DispatchStoreResult<()>,
    {
        platform::claim_pending_unspawned(self, run, launch_hash, child_process_hash, validate)
    }

    pub(crate) fn activate_pending<F>(
        &self,
        run: &RunId,
        launch_hash: [u8; 32],
        child_process_hash: [u8; 32],
        activate: F,
    ) -> DispatchStoreResult<DispatchRecord>
    where
        F: FnOnce(
            &PendingDispatch,
            i64,
            &LockedDispatchRun<'_>,
        ) -> DispatchStoreResult<DispatchRecord>,
    {
        platform::activate_pending(self, run, launch_hash, child_process_hash, activate)
    }

    pub fn reconcile_unspawned(&self, run: &RunId) -> DispatchStoreResult<usize> {
        platform::reconcile_unspawned(self, run, None)
    }

    /// Reconcile every run, including runs that are no longer `executing`.
    /// Broker restart must not depend on active-run discovery because a
    /// claimed launch may belong to a planted, planned, or closing run.
    pub fn reconcile_all_unspawned(&self) -> DispatchStoreResult<usize> {
        let root = platform::open_runs_root(self)?;
        let mut runs = Vec::new();
        for name in platform::run_names(self, &root)? {
            if let Ok(run) = RunId::new(&name) {
                runs.push(run);
            }
        }
        runs.sort();
        runs.dedup();
        let mut reconciled = 0;
        for run in runs {
            match platform::reconcile_unspawned(self, &run, None) {
                Ok(count) => reconciled += count,
                Err(error) if is_not_found(&error) => {}
                Err(error) => return Err(error),
            }
        }
        Ok(reconciled)
    }

    pub fn reconcile_unspawned_at(&self, run: &RunId, now: i64) -> DispatchStoreResult<usize> {
        platform::reconcile_unspawned(self, run, Some(now))
    }

    pub fn cleanup_pending(
        &self,
        run: &RunId,
        launch_hash: [u8; 32],
        state: PendingLaunchState,
    ) -> DispatchStoreResult<LaunchCleanupResponse> {
        if !state.is_terminal() {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidPending(
                "launch cleanup must be terminal".into(),
            )));
        }
        platform::cleanup_pending(self, run, launch_hash, state)
    }

    /// Publish the filesystem half of a committed SQLite singleton intent.
    /// The final record name is no-clobber and the nonce-keyed preparing file
    /// is fsynced before the method returns.
    pub fn publish_singleton_prepared(
        &self,
        publication: &DispatchSingletonPublication,
    ) -> DispatchStoreResult<()> {
        validate_publication_target(self, publication)?;
        let result = platform::publish_singleton(self, publication);
        if result.is_ok()
            && self.publication_fault == Some(PublicationFault::AfterFilesystemPublish)
        {
            return Err(DispatchStoreError::SingletonPublication {
                nonce: publication.nonce.clone(),
                reason: "injected failure after filesystem publication".into(),
            });
        }
        result
    }

    pub(crate) fn publish_review_replacement_prepared(
        &self,
        publication: &DispatchSingletonPublication,
        registry: &mut Registry,
    ) -> DispatchStoreResult<()> {
        platform::publish_review_replacement_prepared(self, publication, registry)
    }

    /// Replay preparing/published intents after a crash or an interrupted
    /// publication. Filesystem evidence is checked against the SQLite hash;
    /// mismatches are quarantined rather than repaired by guesswork.
    pub fn reconcile_singletons(
        &self,
        registry: &mut Registry,
        now: i64,
    ) -> DispatchStoreResult<ReconciliationReport> {
        let publications = registry
            .list_dispatch_publications()
            .map_err(|error| DispatchStoreError::Reconciliation(error.to_string()))?;
        let mut report = ReconciliationReport::default();
        for publication in publications {
            let durable_now = now.max(publication.prepared_at);
            // The existing Native custody recovery finishes interrupted
            // record/pending quarantine first. The publication path below
            // still accepts only the exact typed fourth-rejection delta.
            if publication.state == SingletonPublicationState::Published
                && let Ok(record) = serde_json::from_str::<DispatchRecord>(&publication.record_json)
            {
                match self.load_review_custody(&record.run, &record.agent_id) {
                    Ok(_) => {}
                    Err(error) if is_not_found(&error) => {}
                    Err(error) => return Err(error),
                }
            }
            let action = match platform::reconcile_singleton(self, &publication, registry) {
                Ok(action) => action,
                Err(error) if is_not_found(&error) => PublicationReconcile::Quarantined(
                    "publication filesystem record is missing".into(),
                ),
                Err(error) => return Err(error),
            };
            match action {
                PublicationReconcile::Published => {
                    registry
                        .transaction_immediate::<_, shepherd::registry::Error, _>(|tx| {
                            tx.mark_dispatch_singleton_published(&publication.nonce, durable_now)
                        })
                        .map_err(|error| DispatchStoreError::Reconciliation(error.to_string()))?;
                    report.published += 1;
                }
                PublicationReconcile::Refreshed => report.refreshed += 1,
                PublicationReconcile::Quarantined(reason) => {
                    if let Err(error) = platform::quarantine_singleton(self, &publication)
                        && !is_not_found(&error)
                    {
                        return Err(error);
                    }
                    registry
                        .transaction_immediate::<_, shepherd::registry::Error, _>(|tx| {
                            tx.quarantine_dispatch_singleton(
                                &publication.nonce,
                                &reason,
                                durable_now,
                            )
                        })
                        .map_err(|error| DispatchStoreError::Reconciliation(error.to_string()))?;
                    report.quarantined += 1;
                }
                PublicationReconcile::Unchanged => report.unchanged += 1,
            }
        }
        Ok(report)
    }

    pub fn publish_root_binding(&self, binding: &RootSessionBinding) -> DispatchStoreResult<()> {
        binding.validate()?;
        let active = self.resolve_active_run()?;
        if binding.run != active {
            return Err(DispatchStoreError::WrongActiveRun {
                supplied: binding.run.clone(),
                active,
            });
        }
        platform::publish_root_binding(self, binding)
    }

    pub fn publish_root_binding_for_run(
        &self,
        binding: &RootSessionBinding,
    ) -> DispatchStoreResult<()> {
        binding.validate()?;
        let state = self.load_run(&binding.run)?;
        if matches!(
            state.status.known(),
            Some(RunStatus::Closed | RunStatus::Closing)
        ) {
            return Err(DispatchStoreError::PendingPath {
                run: binding.run.clone(),
            });
        }
        platform::publish_root_binding(self, binding)
    }

    /// Atomically replace one planning binding with the execution binding for
    /// the same trusted root principal after the selected run is executing.
    /// The platform implementation compares the expected bytes while holding
    /// the run lock, so a concurrent rebind cannot be overwritten.
    pub fn transition_root_binding_to_execution(
        &self,
        expected: &RootSessionBinding,
        replacement: &RootSessionBinding,
    ) -> DispatchStoreResult<()> {
        expected.validate()?;
        replacement.validate()?;
        if expected.schema != replacement.schema
            || expected.project_id != replacement.project_id
            || expected.run != replacement.run
            || expected.harness != replacement.harness
            || expected.session_id != replacement.session_id
            || expected.role != replacement.role
            || expected.project_filesystem_id != replacement.project_filesystem_id
            || !expected.mode.is_planting()
            || !replacement.mode.is_execution()
            || replacement.bound_at <= expected.bound_at
        {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidRecord(
                "root binding mode transition is not monotonic planning to execution for one trusted principal"
                    .into(),
            )));
        }
        platform::transition_root_binding_to_execution(self, expected, replacement)
    }

    pub fn load_active_root_binding(
        &self,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        let active = self.resolve_active_run()?;
        platform::load_root_binding(self, &active, session_id)
    }

    pub fn publish_profile_lease(&self, lease: &ProfileLease) -> DispatchStoreResult<()> {
        lease.validate()?;
        let state = self.load_run(&lease.run)?;
        if matches!(
            state.status.known(),
            Some(RunStatus::Closing | RunStatus::Closed)
        ) {
            return Err(DispatchStoreError::PendingPath {
                run: lease.run.clone(),
            });
        }
        platform::publish_profile_lease(self, lease)
    }

    pub fn load_profile_lease(
        &self,
        run: &RunId,
        session_id: &SessionId,
    ) -> DispatchStoreResult<ProfileLease> {
        platform::load_profile_lease(self, run, session_id)
    }

    pub fn replace_profile_lease(
        &self,
        expected: &ProfileLease,
        replacement: &ProfileLease,
    ) -> DispatchStoreResult<()> {
        expected.validate()?;
        replacement.validate()?;
        if expected.project_id != replacement.project_id
            || expected.project_filesystem_id != replacement.project_filesystem_id
            || expected.root_session_id != replacement.root_session_id
            || expected.run != replacement.run
            || expected.harness != replacement.harness
            || expected.profile != replacement.profile
            || expected.expected_attachment != replacement.expected_attachment
            || expected.entered_at != replacement.entered_at
            || expected.expires_at != replacement.expires_at
        {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidProfile(
                "profile replacement changed immutable lease identity".into(),
            )));
        }
        platform::replace_profile_lease(self, expected, replacement)
    }

    pub fn publish_review_custody(&self, custody: &ReviewCustody) -> DispatchStoreResult<()> {
        custody.validate()?;
        platform::publish_review_custody(self, custody)
    }

    pub fn load_review_custody(
        &self,
        run: &RunId,
        subject: &AgentId,
    ) -> DispatchStoreResult<ReviewCustody> {
        platform::load_review_custody(self, run, subject)
    }

    pub(crate) fn read_review_terminal_snapshot(
        &self,
        run: &RunId,
        subject: &AgentId,
    ) -> DispatchStoreResult<ReviewTerminalSnapshot> {
        platform::read_review_terminal_snapshot(self, run, subject)
    }

    pub fn replace_review_custody(
        &self,
        expected: &ReviewCustody,
        replacement: &ReviewCustody,
    ) -> DispatchStoreResult<()> {
        expected.validate()?;
        replacement.validate()?;
        if expected.project_id != replacement.project_id
            || expected.run != replacement.run
            || expected.root_session_id != replacement.root_session_id
            || expected.subject_agent_id != replacement.subject_agent_id
            || expected.subject_session_id != replacement.subject_session_id
            || expected.subject_role != replacement.subject_role
            || expected.lane != replacement.lane
            || expected.pending_launch_id_hash != replacement.pending_launch_id_hash
            || expected.task_sha256 != replacement.task_sha256
            || expected.task_generation != replacement.task_generation
        {
            return Err(DispatchStoreError::Domain(
                DispatchError::InvalidReviewCustody(
                    "review custody replacement changed immutable subject authority".into(),
                ),
            ));
        }
        platform::replace_review_custody(self, expected, replacement)
    }

    pub fn quarantine_malignant(
        &self,
        expected_custody: Option<&ReviewCustody>,
        expected: &DispatchRecord,
        expected_pending: &PendingDispatch,
        custody: &ReviewCustody,
    ) -> DispatchStoreResult<(DispatchRecord, PendingDispatch)> {
        expected.validate_loaded()?;
        expected_pending.validate()?;
        custody.validate()?;
        if custody.state != ReviewCustodyState::Malignant
            || custody.subject_agent_id != expected.agent_id
            || custody.subject_session_id != expected.session_id
            || custody.root_session_id != expected.root_session_id
            || custody.run != expected.run
            || custody.pending_launch_id_hash != expected_pending.launch_id_hash
            || expected_pending.expected_attachment.agent_id != expected.agent_id
            || expected_pending.expected_child_session_id != expected.session_id
            || expected_pending.launch_state != PendingLaunchState::Active
        {
            return Err(DispatchStoreError::Domain(
                DispatchError::InvalidReviewCustody(
                    "malignant quarantine does not match the subject dispatch".into(),
                ),
            ));
        }
        if let Some(expected_custody) = expected_custody {
            expected_custody.validate()?;
            if expected_custody.state != ReviewCustodyState::Active
                || expected_custody.subject_agent_id != custody.subject_agent_id
                || expected_custody.pending_launch_id_hash != custody.pending_launch_id_hash
            {
                return Err(DispatchStoreError::Domain(
                    DispatchError::InvalidReviewCustody(
                        "malignant quarantine changed its active custody identity".into(),
                    ),
                ));
            }
        }
        platform::quarantine_malignant(self, expected_custody, expected, expected_pending, custody)
    }

    pub fn publish_skill_use(&self, challenge: &SkillUseChallenge) -> DispatchStoreResult<()> {
        challenge.validate()?;
        let state = self.load_run(&challenge.run)?;
        if matches!(
            state.status.known(),
            Some(RunStatus::Closing | RunStatus::Closed)
        ) {
            return Err(DispatchStoreError::PendingPath {
                run: challenge.run.clone(),
            });
        }
        platform::publish_skill_use(self, challenge)
    }

    pub fn load_skill_use(
        &self,
        run: &RunId,
        dispatch_id: &AgentId,
        skill: &str,
    ) -> DispatchStoreResult<SkillUseChallenge> {
        platform::load_skill_use(self, run, SkillUseKey::Child(dispatch_id), skill)
    }

    /// Select a root receipt solely from its observed Native authority. No
    /// caller-selected filename or digest can name a different principal.
    pub fn load_root_skill_use(
        &self,
        run: &RunId,
        authority: &SkillUseRootAuthority,
        skill: &str,
    ) -> DispatchStoreResult<SkillUseChallenge> {
        authority.validate()?;
        platform::load_skill_use(self, run, SkillUseKey::Root(authority), skill)
    }

    pub fn replace_skill_use(
        &self,
        expected: &SkillUseChallenge,
        replacement: &SkillUseChallenge,
    ) -> DispatchStoreResult<()> {
        expected.validate()?;
        replacement.validate()?;
        if expected.state != SkillUseState::Pending
            || !matches!(
                replacement.state,
                SkillUseState::Attested | SkillUseState::Expired
            )
        {
            return Err(DispatchStoreError::Domain(DispatchError::SkillUseReplay));
        }
        if expected.project_id != replacement.project_id
            || expected.run != replacement.run
            || expected.dispatch_id != replacement.dispatch_id
            || expected.root_authority != replacement.root_authority
            || expected.session_id != replacement.session_id
            || expected.target != replacement.target
            || expected.role != replacement.role
            || expected.startup_skill != replacement.startup_skill
            || expected.skill != replacement.skill
            || expected.stage != replacement.stage
            || expected.installed_carrier_path != replacement.installed_carrier_path
            || expected.candidate_sha256 != replacement.candidate_sha256
            || expected.carrier_sha256 != replacement.carrier_sha256
            || expected.compiler_tree_sha256 != replacement.compiler_tree_sha256
            || expected.skill_bundle_sha256 != replacement.skill_bundle_sha256
            || expected.nonce_sha256 != replacement.nonce_sha256
            || expected.prepared_at != replacement.prepared_at
            || expected.expires_at != replacement.expires_at
        {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidSkillUse(
                "skill-use replacement changed immutable challenge identity".into(),
            )));
        }
        platform::replace_skill_use(self, expected, replacement)
    }

    pub fn load_active(&self, agent_id: &AgentId) -> DispatchStoreResult<DispatchRecord> {
        let active = self.resolve_active_run()?;
        platform::load(self, &active, agent_id)
    }

    /// Enumerate the native child records for one explicitly named run.
    /// Result paths are never accepted from this enumeration: callers derive
    /// them from each validated record's `result_artifact` field.
    pub fn list_for_run(&self, run: &RunId) -> DispatchStoreResult<Vec<DispatchRecord>> {
        Ok(platform::load_inventory(self, run)?.records)
    }

    /// Reopen one selected run's authority through the caller's existing
    /// anchored lock. The callback cannot retain this borrowed authority.
    pub(crate) fn with_run_access<T>(
        &self,
        run: &RunId,
        access: &crate::run_store::RunAccess<'_>,
        operation: impl FnOnce(&LockedDispatchRun<'_>) -> DispatchStoreResult<T>,
    ) -> DispatchStoreResult<T> {
        platform::with_run_access(self, run, access, operation)
    }

    /// Read one native-issued completion artifact through the run-root walk.
    /// The caller supplies only the allocated relative reference returned in a
    /// dispatch record; path text never selects a different run or file class.
    pub fn read_artifact(&self, run: &RunId, reference: &str) -> DispatchStoreResult<Vec<u8>> {
        validate_artifact_reference(reference)?;
        platform::read_artifact(self, run, reference)
    }

    pub fn resolve_active_identity(
        &self,
        native: &NativeIdentity,
    ) -> DispatchStoreResult<IdentityResolution> {
        self.resolve_active_identity_with_record(native)
            .map(|(resolution, _)| resolution)
    }

    pub(crate) fn resolve_active_identity_with_record(
        &self,
        native: &NativeIdentity,
    ) -> DispatchStoreResult<(IdentityResolution, Option<DispatchRecord>)> {
        let active = self.resolve_active_run()?;
        if native.run != active {
            return Err(DispatchStoreError::WrongActiveRun {
                supplied: native.run.clone(),
                active,
            });
        }
        let record = match &native.agent_id {
            Some(agent_id) => Some(platform::load(self, &active, agent_id)?),
            None => None,
        };
        let resolution = resolve_native_identity(record.as_ref(), native)?;
        Ok((resolution, record))
    }

    pub(crate) fn resolve_identity_for_run_with_record(
        &self,
        native: &NativeIdentity,
    ) -> DispatchStoreResult<(IdentityResolution, Option<DispatchRecord>)> {
        let record = match &native.agent_id {
            Some(agent_id) => Some(platform::load(self, &native.run, agent_id)?),
            None => None,
        };
        let resolution = resolve_native_identity(record.as_ref(), native)?;
        Ok((resolution, record))
    }

    pub fn stop_active(&self, request: StopRequest) -> DispatchStoreResult<DispatchRecord> {
        let active = self.resolve_active_run()?;
        platform::stop(self, &active, request)
    }

    pub fn stop_active_verified(
        &self,
        native: &NativeIdentity,
        request: StopRequest,
    ) -> DispatchStoreResult<DispatchRecord> {
        let active = self.resolve_active_run()?;
        if native.run != active {
            return Err(DispatchStoreError::WrongActiveRun {
                supplied: native.run.clone(),
                active,
            });
        }
        platform::stop_verified(self, &native.run, native, request)
    }

    pub fn stop_verified_for_run(
        &self,
        native: &NativeIdentity,
        request: StopRequest,
    ) -> DispatchStoreResult<DispatchRecord> {
        let state = self.load_run(&native.run)?;
        if state.status.is(RunStatus::Closed) {
            return Err(DispatchStoreError::PendingPath {
                run: native.run.clone(),
            });
        }
        platform::stop_verified(self, &native.run, native, request)
    }

    fn record_path(&self, run: &RunId, agent_id: &AgentId) -> PathBuf {
        self.runs_root
            .join(run.as_str())
            .join("dispatch")
            .join(format!("{}.json", agent_id.as_str()))
    }

    fn root_binding_path(&self, run: &RunId, session_id: &SessionId) -> PathBuf {
        self.runs_root
            .join(run.as_str())
            .join("dispatch")
            .join(root_binding_name(session_id))
    }

    fn current_root_binding_path(&self, session_id: &SessionId) -> PathBuf {
        self.runs_root.join(root_binding_name(session_id))
    }

    fn profile_lease_path(&self, run: &RunId, session_id: &SessionId) -> PathBuf {
        self.runs_root
            .join(run.as_str())
            .join("dispatch")
            .join(profile_lease_name(session_id))
    }

    fn review_custody_path(&self, run: &RunId, subject: &AgentId) -> PathBuf {
        self.runs_root
            .join(run.as_str())
            .join("dispatch")
            .join(review_custody_name(subject))
    }

    #[cfg(not(unix))]
    fn skill_use_path(&self, challenge: &SkillUseChallenge) -> DispatchStoreResult<PathBuf> {
        Ok(self
            .runs_root
            .join(challenge.run.as_str())
            .join("dispatch")
            .join(skill_use_name(
                SkillUseKey::from_challenge(challenge)?,
                &challenge.skill,
            )?))
    }
}

fn publish_review_replacement_at(
    authority: &LockedDispatchRun<'_>,
    publication: &DispatchSingletonPublication,
    registry: &mut Registry,
) -> DispatchStoreResult<()> {
    let target = AgentId::new(&publication.claim.agent_id)?;
    let record = platform::read_locked_record(authority, &target)?;
    let inventory = authority.inventory()?;
    let mut candidates = inventory
        .pending
        .into_iter()
        .filter(|pending| pending.expected_attachment.agent_id == target);
    let pending = candidates.next().ok_or_else(|| {
        DispatchStoreError::Reconciliation(
            "replacement publication has no Native pending record".into(),
        )
    })?;
    let source_id = pending.replaces_agent_id.as_ref().ok_or_else(|| {
        DispatchStoreError::Reconciliation(
            "replacement publication has no exact source lineage".into(),
        )
    })?;
    let source = authority.review_terminal_snapshot(source_id)?;
    let mut bytes = serde_json::to_vec(&record)
        .map_err(|error| DispatchError::InvalidRecord(error.to_string()))?;
    bytes.push(b'\n');
    let attachment_nonce: String = pending
        .nonce_sha256
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect();
    let skill_digest: String = pending
        .expected_attachment
        .skill_bundle_sha256
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect();
    if candidates.next().is_some()
        || pending.launch_state != PendingLaunchState::Active
        || record.state != DispatchState::Active
        || publication.record_json.as_bytes() != bytes
        || pending.project_id != record.project_id
        || pending.run != record.run
        || pending.root_session_id != record.root_session_id
        || pending.role != record.role
        || pending.lane != record.lane
        || pending.expected_child_session_id != record.session_id
        || pending.expected_attachment.target != record.harness
        || pending.activated_at != Some(record.started_at)
        || record.lease_expires_at != pending.expires_at
        || record.parent_agent_id.is_some()
        || record.result_artifact.as_deref() != Some(pending.result_artifact.as_str())
        || record.attachment_nonce.as_deref() != Some(attachment_nonce.as_str())
        || record.startup_attachment.as_ref().is_none_or(|attachment| {
            attachment.skill != pending.expected_attachment.startup_skill
                || attachment.bundle_digest != skill_digest
        })
        || pending.caller_role != shepherd::dispatch::Role::Shepherd
        || pending.parent_dispatch_id.is_some()
        || source.root.role != shepherd::dispatch::Role::Shepherd
        || source.root.project_id != record.project_id
        || source.root.run != record.run
        || source.root.session_id != record.root_session_id
        || !crate::dispatch_service::same_replacement_contract(&source.pending, &pending)
    {
        return Err(DispatchStoreError::Reconciliation(
            "replacement publication requires the exact durable Native activation and authorized lineage".into(),
        ));
    }
    registry
        .transaction_immediate::<_, shepherd::registry::Error, _>(|tx| {
            tx.publish_review_replacement_singleton(
                publication,
                &source.subject,
                &source.pending,
                &source.custody,
                record.started_at,
            )
        })
        .map_err(|error| DispatchStoreError::Reconciliation(error.to_string()))
}

fn finish_preparing_review_replacement(
    authority: &LockedDispatchRun<'_>,
    publication: &DispatchSingletonPublication,
    registry: &mut Registry,
) -> DispatchStoreResult<()> {
    let current = registry
        .load_dispatch_singleton(
            &publication.project_id,
            &publication.run_id,
            &publication.role,
            &publication.lane_key,
        )
        .map_err(|error| DispatchStoreError::Reconciliation(error.to_string()))?;
    if current
        .as_ref()
        .is_none_or(|claim| claim.publication_nonce.as_deref() != Some(publication.nonce.as_str()))
    {
        // An unpointed preparing intent must prove the full review replacement
        // seam. It cannot become an ordinary unclaimed singleton on replay.
        publish_review_replacement_at(authority, publication, registry)?;
    }
    Ok(())
}

fn reconcile_review_terminal_publication(
    authority: &LockedDispatchRun<'_>,
    publication: &DispatchSingletonPublication,
    bytes: Option<&[u8]>,
    registry: &mut Registry,
) -> DispatchStoreResult<bool> {
    let Some(bytes) = bytes else { return Ok(false) };
    let Ok(record_json) = std::str::from_utf8(bytes) else {
        return Ok(false);
    };
    let agent = AgentId::new(&publication.claim.agent_id)?;
    let snapshot = match authority.review_terminal_snapshot(&agent) {
        Ok(snapshot) => snapshot,
        Err(error) if is_not_found(&error) => return Ok(false),
        Err(
            DispatchStoreError::Domain(_)
            | DispatchStoreError::UnknownRecord { .. }
            | DispatchStoreError::PendingNotFound { .. }
            | DispatchStoreError::PendingPath { .. },
        ) => return Ok(false),
        Err(error) => return Err(error),
    };
    if snapshot.root.project_id != snapshot.subject.project_id
        || snapshot.root.run != snapshot.subject.run
        || snapshot.root.session_id != snapshot.subject.root_session_id
        || snapshot.root.role != shepherd::dispatch::Role::Shepherd
    {
        return Ok(false);
    }
    match registry.transaction_immediate::<_, shepherd::registry::Error, _>(|tx| {
        tx.refresh_review_terminal_singleton(
            publication,
            record_json,
            &snapshot.pending,
            &snapshot.custody,
        )
    }) {
        Ok(()) => Ok(true),
        Err(shepherd::registry::Error::InvalidSingletonPublication(_)) => Ok(false),
        Err(error) => Err(DispatchStoreError::Reconciliation(error.to_string())),
    }
}

fn root_binding_name(session_id: &SessionId) -> String {
    format!(".root-session.{}.json", session_id.as_str())
}

fn root_renewal_name(session_id: &SessionId) -> String {
    format!(".root-renewal.{}.json", session_id.as_str())
}

fn profile_lease_name(session_id: &SessionId) -> String {
    format!(".profile.{}.json", session_id.as_str())
}

fn review_custody_name(subject: &AgentId) -> String {
    format!(".review-custody.{}.json", subject.as_str())
}

#[derive(Clone, Copy)]
enum SkillUseKey<'a> {
    Child(&'a AgentId),
    Root(&'a SkillUseRootAuthority),
}

impl<'a> SkillUseKey<'a> {
    fn from_challenge(challenge: &'a SkillUseChallenge) -> DispatchStoreResult<Self> {
        match (&challenge.dispatch_id, &challenge.root_authority) {
            (Some(id), None) => Ok(Self::Child(id)),
            (None, Some(authority)) => Ok(Self::Root(authority)),
            _ => Err(DispatchStoreError::Domain(DispatchError::InvalidSkillUse(
                "skill-use path requires exactly one Native principal".into(),
            ))),
        }
    }
}

fn skill_use_name(key: SkillUseKey<'_>, skill: &str) -> DispatchStoreResult<String> {
    if !matches!(skill, "debugging" | "verification") {
        return Err(DispatchStoreError::Domain(DispatchError::InvalidSkillUse(
            "skill-use path requires a closed Native skill name".into(),
        )));
    }
    match key {
        SkillUseKey::Child(dispatch_id) => Ok(format!(
            ".skill-use.{}.{}.json",
            dispatch_id.as_str(),
            skill
        )),
        SkillUseKey::Root(authority) => {
            use sha2::{Digest, Sha256};
            authority.validate()?;
            let bytes = serde_json::to_vec(authority).map_err(|error| {
                DispatchStoreError::Domain(DispatchError::InvalidSkillUse(error.to_string()))
            })?;
            let digest = Sha256::digest(bytes);
            let digest: String = digest.iter().map(|byte| format!("{byte:02x}")).collect();
            Ok(format!(
                ".root-skill-use.{}.{digest}.{skill}.json",
                authority.binding.session_id.as_str(),
            ))
        }
    }
}

fn pending_name(launch_id_hash: [u8; 32]) -> String {
    let mut name = String::from("pending-");
    for byte in launch_id_hash {
        name.push_str(&format!("{byte:02x}"));
    }
    name.push_str(".json");
    name
}

enum InventoryEntry {
    Record(AgentId),
    Pending([u8; 32]),
}

fn inventory_entry(name: &str) -> DispatchStoreResult<Option<InventoryEntry>> {
    if name.starts_with('.') || !name.ends_with(".json") {
        return Ok(None);
    }
    let stem = name.strip_suffix(".json").expect("checked suffix");
    if let Some(hash) = stem.strip_prefix("pending-") {
        return parse_hash_name(hash)
            .map(InventoryEntry::Pending)
            .map(Some)
            .map_err(|()| {
                DispatchStoreError::Domain(DispatchError::InvalidBudget(
                    "malformed native pending filename in dispatch inventory".into(),
                ))
            });
    }
    Ok(Some(InventoryEntry::Record(AgentId::new(stem)?)))
}

fn sort_inventory(inventory: &mut DispatchInventory) {
    inventory
        .records
        .sort_by(|left, right| left.agent_id.cmp(&right.agent_id));
    inventory
        .pending
        .sort_by_key(|pending| pending.launch_id_hash);
}

#[cfg(test)]
mod budget_tests {
    use super::*;
    use crate::RunStore;

    fn fixture(label: &str) -> PathBuf {
        let path = std::fs::canonicalize(std::env::temp_dir())
            .unwrap()
            .join(format!(
                "shepherd-locked-budget-{label}-{}-{}",
                std::process::id(),
                uuid::Uuid::now_v7()
            ));
        for run in ["v657", "v658"] {
            let directory = path.join(run);
            std::fs::create_dir_all(directory.join("dispatch")).unwrap();
            let state: RunState =
                serde_json::from_value(serde_json::json!({"run":run,"status":"executing"}))
                    .unwrap();
            // Native initialization creates the persistent run lock before
            // parallel dispatch writers can observe the executing state.
            RunStore::new(directory.join("run.json"))
                .initialize(&state)
                .unwrap();
        }
        std::fs::canonicalize(path).unwrap()
    }

    #[test]
    fn held_run_access_enumerates_without_reentrant_lock_or_cross_run_authority() {
        let root = fixture("held-access");
        let run = RunId::new("v657").unwrap();
        let other = RunId::new("v658").unwrap();
        let store = DispatchStore::with_timeout(&root, Duration::from_millis(20));
        RunStore::with_timeout(root.join("v657/run.json"), Duration::from_millis(20))
            .update_with_access(|_, access| {
                store
                    .with_run_access(&run, access, |authority| {
                        assert!(authority.inventory()?.records.is_empty());
                        assert!(authority.inventory()?.pending.is_empty());
                        Ok(())
                    })
                    .unwrap();
                assert!(
                    store
                        .with_run_access(&other, access, |authority| authority.inventory())
                        .is_err()
                );
                Ok(())
            })
            .unwrap();
        std::fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn concurrent_preparers_cannot_both_spend_the_last_native_slot() {
        use std::sync::{Arc, Barrier};
        let root = fixture("concurrent");
        let barrier = Arc::new(Barrier::new(3));
        let outcomes = std::thread::scope(|scope| {
            let handles = [1, 2].map(|marker| {
                let barrier = Arc::clone(&barrier);
                let store = DispatchStore::new(&root);
                scope.spawn(move || {
                    let pending = crate::dispatch_budget::tests::pending(marker);
                    barrier.wait();
                    store.publish_pending_bounded(&pending, 60_000, |candidate, now, authority| {
                        let first = authority.inventory()?;
                        let repeated = authority.inventory()?;
                        assert_eq!(
                            first.pending, repeated.pending,
                            "directory offsets are not shared"
                        );
                        crate::dispatch_budget::measure(candidate, &first, 1, None, now)?
                            .authorize_next()?;
                        Ok(())
                    })
                })
            });
            barrier.wait();
            handles.map(|handle| handle.join().unwrap())
        });
        assert_eq!(outcomes.iter().filter(|value| value.is_ok()).count(), 1);
        let denied = outcomes.into_iter().find_map(Result::err).unwrap();
        assert!(
            matches!(
                denied,
                DispatchStoreError::Domain(DispatchError::HarnessLimit {
                    limit: 1,
                    observed: 2,
                    ..
                })
            ),
            "{denied}"
        );
        std::fs::remove_dir_all(root).unwrap();
    }
}

fn parse_hash_name(value: &str) -> Result<[u8; 32], ()> {
    if value.len() != 64 || !value.is_ascii() {
        return Err(());
    }
    let mut output = [0_u8; 32];
    for (index, pair) in value.as_bytes().as_chunks::<2>().0.iter().enumerate() {
        let high = hex_digit(pair[0]).ok_or(())?;
        let low = hex_digit(pair[1]).ok_or(())?;
        output[index] = high << 4 | low;
    }
    Ok(output)
}

fn hex_digit(value: u8) -> Option<u8> {
    match value {
        b'0'..=b'9' => Some(value - b'0'),
        b'a'..=b'f' => Some(value - b'a' + 10),
        _ => None,
    }
}

fn publication_target_name(
    publication: &DispatchSingletonPublication,
) -> DispatchStoreResult<String> {
    let parts: Vec<_> = publication.record_path.split('/').collect();
    if parts.len() == 3 && parts[1] == "dispatch" && parts[2].ends_with(".json") {
        Ok(parts[2].to_owned())
    } else {
        Err(DispatchStoreError::SingletonPublication {
            nonce: publication.nonce.clone(),
            reason: format!(
                "invalid canonical record path `{}`",
                publication.record_path
            ),
        })
    }
}

fn preparing_name(nonce: &str) -> String {
    format!(".singleton.{nonce}.preparing")
}

fn validate_publication_target(
    store: &DispatchStore,
    publication: &DispatchSingletonPublication,
) -> DispatchStoreResult<()> {
    let parts: Vec<_> = publication.record_path.split('/').collect();
    let valid = parts.len() == 3
        && parts[0] == publication.run_id
        && parts[1] == "dispatch"
        && parts[2].strip_suffix(".json").is_some_and(|agent| {
            !agent.is_empty()
                && agent.bytes().all(|byte| {
                    byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b':' | b'-')
                })
        })
        && !publication.record_path.contains(['\\', '\0'])
        && !publication.record_path.chars().any(char::is_control)
        && store
            .runs_root
            .join(&publication.record_path)
            .starts_with(&store.runs_root);
    if valid {
        Ok(())
    } else {
        Err(DispatchStoreError::SingletonPublication {
            nonce: publication.nonce.clone(),
            reason: format!(
                "invalid canonical record path `{}`",
                publication.record_path
            ),
        })
    }
}

/// Resolve the single `status == "executing"` run beneath the runs root.
///
/// Enumeration is deliberately tolerant of what it finds. `.shepherd/runs/`
/// accumulates directories that are not runs -- a legacy namespace holding
/// only `plan.md`, an empty tree, a scratch tree carrying `dispatch/` and no
/// `run.json` -- and that set is unbounded, so repairing one only reveals the
/// next. A namespace whose `run.json` is *absent* is therefore passed over
/// exactly as a name that does not parse as a `RunId` already is.
///
/// Absence is the only tolerated failure. A corrupt document still raises
/// `InvalidRunDocument` and a linked or non-regular one still raises
/// `UnsafePath`: swallowing those would trade a loud abort for a silent
/// `NoActiveRun` and delete the diagnosis, which is the same deadlock with
/// less to go on (#330).
///
/// Everything above the anchoring is platform-free and lives here once. The
/// `platform` twins contribute only anchored root enumeration and document reads.
fn resolve_active_run(store: &DispatchStore) -> DispatchStoreResult<RunId> {
    let root = platform::open_runs_root(store)?;
    let mut names = Vec::new();
    for name in platform::run_names(store, &root)? {
        if name == "." || name == ".." {
            continue;
        }
        if let Ok(run) = RunId::new(&name) {
            names.push(run);
        }
    }
    names.sort();
    names.dedup();

    let mut active = Vec::new();
    for run in names {
        let state = match platform::read_run_document(store, &root, &run) {
            Ok(state) => state,
            Err(error) if is_not_found(&error) => continue,
            Err(error) => return Err(error),
        };
        if state.status.is(RunStatus::Executing) {
            active.push(run);
        }
    }
    match active.len() {
        0 => Err(DispatchStoreError::NoActiveRun),
        1 => Ok(active.remove(0)),
        _ => Err(DispatchStoreError::AmbiguousActiveRuns { runs: active }),
    }
}

/// Whether `error` is the filesystem reporting that a name does not exist.
///
/// `unsafe_path_or_io` sends `ELOOP` and `ENOTDIR` to `UnsafePath`, so what
/// reaches `Io { NotFound }` really is "no such entry" and nothing else.
fn current_time_millis() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .ok()
        .and_then(|duration| i64::try_from(duration.as_millis()).ok())
        .unwrap_or(i64::MAX)
}

fn refresh_pending_lease(
    pending: &PendingDispatch,
    lease_ms: Option<u64>,
) -> DispatchStoreResult<PendingDispatch> {
    let Some(lease_ms) = lease_ms else {
        return Ok(pending.clone());
    };
    if lease_ms == 0 || lease_ms > 86_400_000 {
        return Err(DispatchStoreError::Domain(DispatchError::InvalidPending(
            "pending lease is outside the native bound".into(),
        )));
    }
    let lease_ms = i64::try_from(lease_ms).map_err(|_| {
        DispatchStoreError::Domain(DispatchError::InvalidPending(
            "pending lease overflows i64".into(),
        ))
    })?;
    let now = current_time_millis();
    let mut refreshed = pending.clone();
    refreshed.expires_at = now.checked_add(lease_ms).ok_or_else(|| {
        DispatchStoreError::Domain(DispatchError::InvalidPending(
            "pending lease overflows time".into(),
        ))
    })?;
    refreshed.validate()?;
    Ok(refreshed)
}

fn validate_artifact_reference(reference: &str) -> DispatchStoreResult<()> {
    if reference.is_empty()
        || reference.len() > 512
        || reference.starts_with('/')
        || reference.contains(['\\', '\0'])
        || reference.chars().any(char::is_control)
        || reference
            .split('/')
            .any(|part| part.is_empty() || part == "." || part == "..")
    {
        return Err(DispatchStoreError::UnsafePath {
            path: PathBuf::from(reference),
        });
    }
    Ok(())
}

fn is_not_found(error: &DispatchStoreError) -> bool {
    matches!(
        error,
        DispatchStoreError::Io { source, .. }
            if source.kind() == std::io::ErrorKind::NotFound
    )
}

fn decode_root_binding(
    bytes: &[u8],
    run: &RunId,
    session_id: &SessionId,
) -> DispatchStoreResult<RootSessionBinding> {
    let binding: RootSessionBinding = serde_json::from_slice(bytes).map_err(|error| {
        DispatchStoreError::Domain(DispatchError::InvalidRecord(error.to_string()))
    })?;
    binding.validate()?;
    if &binding.run != run || &binding.session_id != session_id {
        return Err(DispatchStoreError::Domain(DispatchError::InvalidRecord(
            "root binding identity does not match its canonical path".into(),
        )));
    }
    Ok(binding)
}

fn decode_current_root_binding(
    bytes: &[u8],
    session_id: &SessionId,
) -> DispatchStoreResult<RootSessionBinding> {
    let binding: RootSessionBinding = serde_json::from_slice(bytes).map_err(|error| {
        DispatchStoreError::Domain(DispatchError::InvalidRecord(error.to_string()))
    })?;
    binding.validate()?;
    if &binding.session_id != session_id {
        return Err(DispatchStoreError::Domain(DispatchError::InvalidRecord(
            "current root binding session does not match its canonical path".into(),
        )));
    }
    Ok(binding)
}

fn validate_current_binding_transition(
    current: &RootSessionBinding,
    replacement: &RootSessionBinding,
) -> DispatchStoreResult<()> {
    let same_principal = current.schema == replacement.schema
        && current.project_id == replacement.project_id
        && current.harness == replacement.harness
        && current.session_id == replacement.session_id
        && current.role == replacement.role
        && current.project_filesystem_id == replacement.project_filesystem_id;
    let same_run_transition = current.run == replacement.run
        && (current.mode == replacement.mode
            || (current.mode.is_planting() && replacement.mode.is_execution()));
    let next_run_transition = current.run != replacement.run && replacement.mode.is_planting();
    if !same_principal
        || replacement.bound_at <= current.bound_at
        || !(same_run_transition || next_run_transition)
    {
        return Err(DispatchStoreError::Domain(DispatchError::InvalidRecord(
            "current root binding transition is not a monotonic same-principal continuation".into(),
        )));
    }
    Ok(())
}

fn decode_profile_lease(
    bytes: &[u8],
    run: &RunId,
    session_id: &SessionId,
) -> DispatchStoreResult<ProfileLease> {
    let lease: ProfileLease = serde_json::from_slice(bytes).map_err(|error| {
        DispatchStoreError::Domain(DispatchError::InvalidProfile(error.to_string()))
    })?;
    lease.validate()?;
    if &lease.run != run || &lease.root_session_id != session_id {
        return Err(DispatchStoreError::Domain(DispatchError::InvalidProfile(
            "profile lease identity does not match its canonical path".into(),
        )));
    }
    Ok(lease)
}

fn decode_review_custody(
    bytes: &[u8],
    run: &RunId,
    subject: &AgentId,
) -> DispatchStoreResult<ReviewCustody> {
    let custody: ReviewCustody = serde_json::from_slice(bytes).map_err(|error| {
        DispatchStoreError::Domain(DispatchError::InvalidReviewCustody(error.to_string()))
    })?;
    custody.validate()?;
    if &custody.run != run || &custody.subject_agent_id != subject {
        return Err(DispatchStoreError::Domain(
            DispatchError::InvalidReviewCustody(
                "review custody identity does not match its canonical path".into(),
            ),
        ));
    }
    Ok(custody)
}

fn decode_skill_use(
    bytes: &[u8],
    run: &RunId,
    key: SkillUseKey<'_>,
    skill: &str,
) -> DispatchStoreResult<SkillUseChallenge> {
    let challenge: SkillUseChallenge = serde_json::from_slice(bytes).map_err(|error| {
        DispatchStoreError::Domain(DispatchError::InvalidSkillUse(error.to_string()))
    })?;
    challenge.validate()?;
    let principal_matches = match key {
        SkillUseKey::Child(id) => {
            challenge.dispatch_id.as_ref() == Some(id) && challenge.root_authority.is_none()
        }
        SkillUseKey::Root(authority) => {
            challenge.dispatch_id.is_none() && challenge.root_authority.as_ref() == Some(authority)
        }
    };
    if &challenge.run != run || !principal_matches || challenge.skill != skill {
        return Err(DispatchStoreError::Domain(DispatchError::InvalidSkillUse(
            "skill-use identity does not match its canonical path".into(),
        )));
    }
    Ok(challenge)
}

#[cfg(unix)]
mod platform {
    use std::fs::TryLockError;
    use std::os::fd::OwnedFd;

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

    use super::*;

    const MAX_RECORD_BYTES: u64 = 1_048_576;
    const LOCK_RETRY_INTERVAL: Duration = Duration::from_millis(10);
    const MAX_TEMP_ATTEMPTS: u32 = 100;
    static TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(0);

    pub(super) fn publish_singleton(
        store: &DispatchStore,
        publication: &DispatchSingletonPublication,
    ) -> DispatchStoreResult<()> {
        let root = open_runs_root(store)?;
        let run = RunId::new(publication.run_id.clone()).map_err(DispatchStoreError::Domain)?;
        let run_fd = open_run_dir(store, &root, &run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, &run, true)?;
        let _lock = acquire_lock(store, &dispatch_fd, &run)?;
        let target = publication_target_name(publication)?;
        let temp = preparing_name(&publication.nonce);
        let bytes = publication.record_json.as_bytes();
        ensure_preparing_file(store, &dispatch_fd, &run, &temp, bytes)?;
        if store.publication_fault == Some(PublicationFault::AfterPreparing) {
            return Err(DispatchStoreError::SingletonPublication {
                nonce: publication.nonce.clone(),
                reason: "injected failure after preparing filesystem bytes".into(),
            });
        }
        publish_final_no_clobber(store, &dispatch_fd, &run, &temp, &target, bytes)
    }

    pub(super) fn publish_review_replacement_prepared(
        store: &DispatchStore,
        publication: &DispatchSingletonPublication,
        registry: &mut Registry,
    ) -> DispatchStoreResult<()> {
        let root = open_runs_root(store)?;
        let run = RunId::new(&publication.run_id)?;
        let run_fd = open_run_dir(store, &root, &run)?;
        let dispatch = open_dispatch_dir(store, &run_fd, &run, false)?;
        let _lock = acquire_lock(store, &run_fd, &run)?;
        if read_optional_at(
            &dispatch,
            &publication_target_name(publication)?,
            MAX_RECORD_BYTES,
        )?
        .as_deref()
            != Some(publication.record_json.as_bytes())
        {
            return Err(DispatchStoreError::Reconciliation(
                "replacement filesystem bytes differ from the prepared publication".into(),
            ));
        }
        publish_review_replacement_at(
            &LockedDispatchRun {
                store,
                run: &run,
                dispatch: &dispatch,
            },
            publication,
            registry,
        )
    }

    pub(super) fn reconcile_singleton(
        store: &DispatchStore,
        publication: &DispatchSingletonPublication,
        registry: &mut Registry,
    ) -> DispatchStoreResult<PublicationReconcile> {
        let root = open_runs_root(store)?;
        let run = RunId::new(publication.run_id.clone()).map_err(DispatchStoreError::Domain)?;
        let run_fd = open_run_dir(store, &root, &run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, &run, false)?;
        let _lock = acquire_lock(store, &run_fd, &run)?;
        let target = publication_target_name(publication)?;
        let temp = preparing_name(&publication.nonce);
        let expected = publication.record_json.as_bytes();
        let target_bytes = read_optional_at(&dispatch_fd, &target, MAX_RECORD_BYTES)?;
        match publication.state {
            SingletonPublicationState::Quarantined => Ok(PublicationReconcile::Unchanged),
            SingletonPublicationState::Published => {
                if target_bytes.as_deref() == Some(expected) {
                    Ok(PublicationReconcile::Unchanged)
                } else if reconcile_review_terminal_publication(
                    &LockedDispatchRun {
                        store,
                        run: &run,
                        dispatch: &dispatch_fd,
                    },
                    publication,
                    target_bytes.as_deref(),
                    registry,
                )? {
                    Ok(PublicationReconcile::Refreshed)
                } else {
                    Ok(PublicationReconcile::Quarantined(
                        "published filesystem bytes are missing or corrupt".into(),
                    ))
                }
            }
            SingletonPublicationState::Preparing => {
                let temp_bytes = read_optional_at(&dispatch_fd, &temp, MAX_RECORD_BYTES)?;
                if target_bytes
                    .as_deref()
                    .is_some_and(|value| value != expected)
                    || temp_bytes.as_deref().is_some_and(|value| value != expected)
                {
                    return Ok(PublicationReconcile::Quarantined(
                        "preparing or published filesystem bytes are corrupt".into(),
                    ));
                }
                if target_bytes.as_deref() == Some(expected) {
                    finish_preparing_review_replacement(
                        &LockedDispatchRun {
                            store,
                            run: &run,
                            dispatch: &dispatch_fd,
                        },
                        publication,
                        registry,
                    )?;
                    remove_optional_at(&dispatch_fd, &temp)?;
                    fs::fsync(&dispatch_fd).map_err(|source| {
                        DispatchStoreError::io(
                            "fsync reconciled dispatch directory",
                            store.runs_root.join(run.as_str()).join("dispatch"),
                            source,
                        )
                    })?;
                    return Ok(PublicationReconcile::Published);
                }
                if temp_bytes.as_deref() == Some(expected) {
                    publish_final_no_clobber(store, &dispatch_fd, &run, &temp, &target, expected)?;
                    finish_preparing_review_replacement(
                        &LockedDispatchRun {
                            store,
                            run: &run,
                            dispatch: &dispatch_fd,
                        },
                        publication,
                        registry,
                    )?;
                    return Ok(PublicationReconcile::Published);
                }
                Ok(PublicationReconcile::Quarantined(
                    "preparing publication has no durable filesystem bytes".into(),
                ))
            }
        }
    }

    pub(super) fn quarantine_singleton(
        store: &DispatchStore,
        publication: &DispatchSingletonPublication,
    ) -> DispatchStoreResult<()> {
        let root = open_runs_root(store)?;
        let run = RunId::new(publication.run_id.clone()).map_err(DispatchStoreError::Domain)?;
        let run_fd = open_run_dir(store, &root, &run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, &run, false)?;
        let _lock = acquire_lock(store, &dispatch_fd, &run)?;
        let quarantine_path = store
            .runs_root
            .join(run.as_str())
            .join("dispatch/quarantine");
        match mkdirat(&dispatch_fd, "quarantine", Mode::RWXU) {
            Ok(()) | Err(rustix::io::Errno::EXIST) => {}
            Err(source) => {
                return Err(DispatchStoreError::io(
                    "create singleton quarantine directory",
                    quarantine_path,
                    source,
                ));
            }
        }
        let quarantine_fd = openat(
            &dispatch_fd,
            "quarantine",
            OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::empty(),
        )
        .map_err(|source| {
            unsafe_path_or_io(
                "open singleton quarantine directory",
                quarantine_path.clone(),
                source,
            )
        })?;
        let temp = preparing_name(&publication.nonce);
        let preparing_quarantine = format!("{}.preparing", publication.nonce);
        let moved_preparing = move_to_quarantine(
            &dispatch_fd,
            &quarantine_fd,
            &temp,
            &preparing_quarantine,
            &quarantine_path,
        )?;
        if !moved_preparing {
            write_quarantine_marker(
                &quarantine_fd,
                &preparing_quarantine,
                publication.record_json.as_bytes(),
                &quarantine_path,
            )?;
        }
        let target = publication_target_name(publication)?;
        let published_quarantine = format!("{}.published", publication.nonce);
        let _ = move_to_quarantine(
            &dispatch_fd,
            &quarantine_fd,
            &target,
            &published_quarantine,
            &quarantine_path,
        )?;
        fs::fsync(&quarantine_fd).map_err(|source| {
            DispatchStoreError::io(
                "fsync singleton quarantine",
                quarantine_path.clone(),
                source,
            )
        })?;
        fs::fsync(&dispatch_fd).map_err(|source| {
            DispatchStoreError::io(
                "fsync singleton dispatch directory",
                store.runs_root.join(run.as_str()).join("dispatch"),
                source,
            )
        })
    }

    fn move_to_quarantine(
        dispatch_fd: &OwnedFd,
        quarantine_fd: &OwnedFd,
        source: &str,
        destination: &str,
        quarantine_path: &Path,
    ) -> DispatchStoreResult<bool> {
        match linkat(
            dispatch_fd,
            source,
            quarantine_fd,
            destination,
            AtFlags::empty(),
        ) {
            Ok(()) => {
                unlinkat(dispatch_fd, source, AtFlags::empty()).map_err(|error| {
                    DispatchStoreError::io(
                        "remove singleton quarantined source",
                        quarantine_path.join(source),
                        error,
                    )
                })?;
                Ok(true)
            }
            Err(error) if error == rustix::io::Errno::NOENT => Ok(false),
            Err(error) if error == rustix::io::Errno::EXIST => {
                unlinkat(dispatch_fd, source, AtFlags::empty()).map_err(|unlink_error| {
                    DispatchStoreError::io(
                        "remove duplicate singleton quarantined source",
                        quarantine_path.join(source),
                        unlink_error,
                    )
                })?;
                Ok(true)
            }
            Err(error) => Err(DispatchStoreError::io(
                "quarantine singleton filesystem record",
                quarantine_path.join(destination),
                error,
            )),
        }
    }

    fn write_quarantine_marker(
        quarantine_fd: &OwnedFd,
        name: &str,
        bytes: &[u8],
        quarantine_path: &Path,
    ) -> DispatchStoreResult<()> {
        match openat(
            quarantine_fd,
            name,
            OFlags::WRONLY | OFlags::CREATE | OFlags::EXCL | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::RUSR | Mode::WUSR,
        ) {
            Ok(fd) => {
                let mut file = File::from(fd);
                file.write_all(bytes).map_err(|error| {
                    DispatchStoreError::io(
                        "write singleton quarantine marker",
                        quarantine_path.join(name),
                        error,
                    )
                })?;
                file.sync_all().map_err(|error| {
                    DispatchStoreError::io(
                        "fsync singleton quarantine marker",
                        quarantine_path.join(name),
                        error,
                    )
                })?;
            }
            Err(error) if error == rustix::io::Errno::EXIST => {}
            Err(error) => {
                return Err(DispatchStoreError::io(
                    "write singleton quarantine marker",
                    quarantine_path.join(name),
                    error,
                ));
            }
        }
        Ok(())
    }

    pub(super) fn publish_root_binding(
        store: &DispatchStore,
        binding: &RootSessionBinding,
    ) -> DispatchStoreResult<()> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, &binding.run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, &binding.run, true)?;
        let _lock = acquire_lock(store, &run_fd, &binding.run)?;
        let name = root_binding_name(&binding.session_id);
        let bytes = encode_document(binding)?;
        publish_no_clobber(store, &dispatch_fd, &binding.run, &name, &bytes)
    }

    pub(super) fn load_current_root_binding(
        store: &DispatchStore,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        let root = open_runs_root(store)?;
        read_current_root_binding_at(store, &root, session_id)
    }

    pub(super) fn root_renewal_intent(
        store: &DispatchStore,
        session: &SessionId,
    ) -> DispatchStoreResult<Option<RootRenewalIntent>> {
        let root = open_runs_root(store)?;
        read_optional_at(&root, &root_renewal_name(session), MAX_RECORD_BYTES)?
            .map(|bytes| decode_root_renewal(&bytes, session))
            .transpose()
    }

    fn ensure_no_root_renewal_at(root: &OwnedFd, session: &SessionId) -> DispatchStoreResult<()> {
        if read_optional_at(root, &root_renewal_name(session), MAX_RECORD_BYTES)?.is_some() {
            return Err(root_renewal_error(
                "root lease renewal is incomplete; repeat the exact explicit bind-root request",
            ));
        }
        Ok(())
    }

    pub(super) fn commit_root_renewal(
        store: &DispatchStore,
        intent: &RootRenewalIntent,
        recover: bool,
    ) -> DispatchStoreResult<RootSessionBinding> {
        let root = open_runs_root(store)?;
        let _current_lock = acquire_current_root_lock(store, &root)?;
        let run = &intent.expected.run;
        let session = &intent.expected.session_id;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch = open_dispatch_dir(store, &run_fd, run, false)?;
        let _run_lock = acquire_lock(store, &run_fd, run)?;
        let name = root_renewal_name(session);
        let existing_intent = read_optional_at(&root, &name, MAX_RECORD_BYTES)?
            .map(|bytes| decode_root_renewal(&bytes, session))
            .transpose()?;
        if (recover && existing_intent.as_ref() != Some(intent))
            || (!recover && existing_intent.is_some())
        {
            return Err(root_renewal_error(
                "root renewal intent changed or requires explicit same-root recovery",
            ));
        }
        let binding = read_root_binding_raw_at(store, &dispatch, run, session)?;
        let index = read_current_root_binding_raw_at(store, &root, session)?;
        let profile = match read_profile_lease_at(store, &dispatch, run, session) {
            Ok(lease) => Some(lease),
            Err(error) if is_not_found(&error) => None,
            Err(error) => return Err(error),
        };
        let state = read_run_document(store, &root, run)?;
        let (run_old, index_old) = validate_root_renewal_observation(
            intent,
            &binding,
            &index,
            profile.as_ref(),
            &state.status,
        )?;
        if !recover && !(run_old && index_old) {
            return Err(root_renewal_error(
                "root renewal expected authority is no longer current",
            ));
        }
        if !recover {
            publish_root_renewal_intent(store, &root, run, &name, &encode_document(intent)?)?;
            store.root_renewal_fault(RootRenewalFault::AfterIntentCommit)?;
        }
        let bytes = encode_document(&intent.replacement)?;
        if run_old {
            replace_existing(store, &dispatch, run, &root_binding_name(session), &bytes)?;
            store.root_renewal_fault(RootRenewalFault::AfterRunBindingCommit)?;
        }
        if index_old {
            replace_existing(store, &root, run, &root_binding_name(session), &bytes)?;
            store.root_renewal_fault(RootRenewalFault::AfterIndexCommit)?;
        }
        if read_root_binding_raw_at(store, &dispatch, run, session)? != intent.replacement
            || read_current_root_binding_raw_at(store, &root, session)? != intent.replacement
        {
            return Err(root_renewal_error(
                "root renewal records changed before commit",
            ));
        }
        remove_optional_at(&root, &name)?;
        fs::fsync(&root).map_err(|error| {
            DispatchStoreError::io(
                "fsync committed root renewal",
                store.runs_root.clone(),
                error,
            )
        })?;
        Ok(intent.replacement.clone())
    }

    fn publish_root_renewal_intent(
        store: &DispatchStore,
        root: &OwnedFd,
        run: &RunId,
        name: &str,
        bytes: &[u8],
    ) -> DispatchStoreResult<()> {
        let (temporary, mut file) = create_temp(store, root, run, name)?;
        file.write_all(bytes).map_err(|error| {
            DispatchStoreError::io(
                "write root renewal intent",
                store.runs_root.join(&temporary),
                error,
            )
        })?;
        file.sync_all().map_err(|error| {
            DispatchStoreError::io(
                "fsync root renewal intent",
                store.runs_root.join(&temporary),
                error,
            )
        })?;
        // A link+unlink publisher has an unrecoverable crash seam: the durable
        // intent can retain two links, which our authority readers must reject.
        // A no-replace rename publishes one complete, single-link inode.
        let published = rename_root_renewal_intent(root, &temporary, name);
        if let Err(error) = published {
            let _ = remove_optional_at(root, &temporary);
            return Err(DispatchStoreError::io(
                "publish root renewal intent",
                store.runs_root.join(name),
                error,
            ));
        }
        store.root_renewal_fault(RootRenewalFault::AfterIntentPublish)?;
        fs::fsync(root).map_err(|error| {
            DispatchStoreError::io(
                "fsync root renewal directory",
                store.runs_root.clone(),
                error,
            )
        })
    }

    fn rename_root_renewal_intent(
        root: &OwnedFd,
        temporary: &str,
        name: &str,
    ) -> std::io::Result<()> {
        #[cfg(any(
            target_vendor = "apple",
            target_os = "linux",
            target_os = "android",
            target_os = "redox"
        ))]
        {
            fs::renameat_with(root, temporary, root, name, fs::RenameFlags::NOREPLACE)
                .map_err(Into::into)
        }
        #[cfg(not(any(
            target_vendor = "apple",
            target_os = "linux",
            target_os = "android",
            target_os = "redox"
        )))]
        {
            let _ = (root, temporary, name);
            Err(std::io::Error::new(
                std::io::ErrorKind::Unsupported,
                "atomic no-replace root renewal publication is unavailable on this platform",
            ))
        }
    }

    fn read_current_root_binding_at(
        store: &DispatchStore,
        root: &OwnedFd,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        ensure_no_root_renewal_at(root, session_id)?;
        read_current_root_binding_raw_at(store, root, session_id)
    }

    fn read_current_root_binding_raw_at(
        store: &DispatchStore,
        root: &OwnedFd,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        let name = root_binding_name(session_id);
        let path = store.current_root_binding_path(session_id);
        let file = open_regular_at(root, &name, &path)?;
        let bytes = read_bounded(file, MAX_RECORD_BYTES).map_err(|source| {
            DispatchStoreError::io("read current root binding", path.clone(), source)
        })?;
        decode_current_root_binding(&bytes, session_id)
    }

    pub(super) fn activate_current_root_binding(
        store: &DispatchStore,
        binding: &RootSessionBinding,
    ) -> DispatchStoreResult<()> {
        let root = open_runs_root(store)?;
        let _lock = acquire_current_root_lock(store, &root)?;
        let name = root_binding_name(&binding.session_id);
        let bytes = encode_document(binding)?;
        match read_current_root_binding_at(store, &root, &binding.session_id) {
            Ok(current) if current == *binding => Ok(()),
            Ok(current) => {
                validate_current_binding_transition(&current, binding)?;
                replace_existing(store, &root, &binding.run, &name, &bytes)
            }
            Err(error) if is_not_found(&error) => {
                publish_no_clobber(store, &root, &binding.run, &name, &bytes)
            }
            Err(error) => Err(error),
        }
    }

    pub(super) fn transition_root_binding_to_execution(
        store: &DispatchStore,
        expected: &RootSessionBinding,
        replacement: &RootSessionBinding,
    ) -> DispatchStoreResult<()> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, &replacement.run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, &replacement.run, false)?;
        let _lock = acquire_lock(store, &run_fd, &replacement.run)?;
        let state = read_run_document(store, &root, &replacement.run)?;
        if !state.status.is(RunStatus::Executing) {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidRecord(
                "root binding cannot enter execution before the selected run is executing".into(),
            )));
        }
        let current = read_root_binding_at(
            store,
            &dispatch_fd,
            &replacement.run,
            &replacement.session_id,
        )?;
        if &current != expected {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidRecord(
                "root binding changed during mode transition".into(),
            )));
        }
        replace_existing(
            store,
            &dispatch_fd,
            &replacement.run,
            &root_binding_name(&replacement.session_id),
            &encode_document(replacement)?,
        )
    }

    pub(super) fn publish_profile_lease(
        store: &DispatchStore,
        lease: &ProfileLease,
    ) -> DispatchStoreResult<()> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, &lease.run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, &lease.run, true)?;
        let _lock = acquire_lock(store, &run_fd, &lease.run)?;
        let name = profile_lease_name(&lease.root_session_id);
        publish_no_clobber(
            store,
            &dispatch_fd,
            &lease.run,
            &name,
            &encode_document(lease)?,
        )
    }

    pub(super) fn replace_profile_lease(
        store: &DispatchStore,
        expected: &ProfileLease,
        replacement: &ProfileLease,
    ) -> DispatchStoreResult<()> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, &replacement.run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, &replacement.run, false)?;
        let _lock = acquire_lock(store, &run_fd, &replacement.run)?;
        let current = read_profile_lease_at(
            store,
            &dispatch_fd,
            &replacement.run,
            &replacement.root_session_id,
        )?;
        if &current != expected {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidProfile(
                "profile lease changed during state transition".into(),
            )));
        }
        replace_existing(
            store,
            &dispatch_fd,
            &replacement.run,
            &profile_lease_name(&replacement.root_session_id),
            &encode_document(replacement)?,
        )
    }

    pub(super) fn publish_review_custody(
        store: &DispatchStore,
        custody: &ReviewCustody,
    ) -> DispatchStoreResult<()> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, &custody.run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, &custody.run, true)?;
        let _lock = acquire_lock(store, &run_fd, &custody.run)?;
        publish_no_clobber(
            store,
            &dispatch_fd,
            &custody.run,
            &review_custody_name(&custody.subject_agent_id),
            &encode_document(custody)?,
        )
    }

    pub(super) fn replace_review_custody(
        store: &DispatchStore,
        expected: &ReviewCustody,
        replacement: &ReviewCustody,
    ) -> DispatchStoreResult<()> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, &replacement.run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, &replacement.run, false)?;
        let _lock = acquire_lock(store, &run_fd, &replacement.run)?;
        let current = read_review_custody_at(
            store,
            &dispatch_fd,
            &replacement.run,
            &replacement.subject_agent_id,
        )?;
        if &current != expected {
            return Err(DispatchStoreError::Domain(
                DispatchError::InvalidReviewCustody(
                    "review custody changed during state transition".into(),
                ),
            ));
        }
        replace_existing(
            store,
            &dispatch_fd,
            &replacement.run,
            &review_custody_name(&replacement.subject_agent_id),
            &encode_document(replacement)?,
        )
    }

    pub(super) fn publish_skill_use(
        store: &DispatchStore,
        challenge: &SkillUseChallenge,
    ) -> DispatchStoreResult<()> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, &challenge.run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, &challenge.run, true)?;
        let _lock = acquire_lock(store, &run_fd, &challenge.run)?;
        let key = SkillUseKey::from_challenge(challenge)?;
        validate_skill_use_root_at(store, &root, &dispatch_fd, &challenge.run, key)?;
        let name = skill_use_name(key, &challenge.skill)?;
        publish_no_clobber(
            store,
            &dispatch_fd,
            &challenge.run,
            &name,
            &encode_document(challenge)?,
        )
    }

    pub(super) fn replace_skill_use(
        store: &DispatchStore,
        expected: &SkillUseChallenge,
        replacement: &SkillUseChallenge,
    ) -> DispatchStoreResult<()> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, &replacement.run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, &replacement.run, false)?;
        let _lock = acquire_lock(store, &run_fd, &replacement.run)?;
        let key = SkillUseKey::from_challenge(replacement)?;
        validate_skill_use_root_at(store, &root, &dispatch_fd, &replacement.run, key)?;
        let current = read_skill_use_at(
            store,
            &dispatch_fd,
            &replacement.run,
            key,
            &replacement.skill,
        )?;
        if &current != expected {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidSkillUse(
                "skill-use challenge changed during state transition".into(),
            )));
        }
        replace_existing(
            store,
            &dispatch_fd,
            &replacement.run,
            &skill_use_name(key, &replacement.skill)?,
            &encode_document(replacement)?,
        )
    }

    pub(super) fn publish_pending<F>(
        store: &DispatchStore,
        pending: &PendingDispatch,
        lease_ms: Option<u64>,
        authorize: F,
    ) -> DispatchStoreResult<PendingDispatch>
    where
        F: FnOnce(&PendingDispatch, i64, &LockedDispatchRun<'_>) -> DispatchStoreResult<()>,
    {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, &pending.run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, &pending.run, true)?;
        let _lock = acquire_lock(store, &run_fd, &pending.run)?;
        let state = store.load_run(&pending.run)?;
        if matches!(
            state.status.known(),
            Some(RunStatus::Closed | RunStatus::Closing)
        ) {
            return Err(DispatchStoreError::PendingPath {
                run: pending.run.clone(),
            });
        }
        if state.status != pending.run_status {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidPending(
                "run state changed during pending publication".into(),
            )));
        }
        let agent_id = &pending.expected_attachment.agent_id;
        match read_record(store, &dispatch_fd, &pending.run, agent_id) {
            Ok(_) => {
                return Err(DispatchStoreError::AlreadyExists {
                    path: store.record_path(&pending.run, agent_id),
                });
            }
            Err(DispatchStoreError::UnknownRecord { reason, .. })
                if reason == "record is absent" => {}
            Err(error) => return Err(error),
        }
        let pending = refresh_pending_lease(pending, lease_ms)?;
        let authority = LockedDispatchRun {
            store,
            run: &pending.run,
            dispatch: &dispatch_fd,
        };
        authorize(&pending, current_time_millis(), &authority)?;
        let name = pending_name(pending.launch_id_hash);
        let bytes = encode_document(&pending)?;
        publish_no_clobber(store, &dispatch_fd, &pending.run, &name, &bytes)?;
        Ok(pending)
    }

    pub(super) fn load_inventory(
        store: &DispatchStore,
        run: &RunId,
    ) -> DispatchStoreResult<DispatchInventory> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, run, false)?;
        let _lock = acquire_lock(store, &run_fd, run)?;
        read_locked_inventory(&LockedDispatchRun {
            store,
            run,
            dispatch: &dispatch_fd,
        })
    }

    pub(super) fn with_run_access<T>(
        store: &DispatchStore,
        run: &RunId,
        access: &crate::run_store::RunAccess<'_>,
        operation: impl FnOnce(&LockedDispatchRun<'_>) -> DispatchStoreResult<T>,
    ) -> DispatchStoreResult<T> {
        let root = open_runs_root(store)?;
        let canonical_run = open_run_dir(store, &root, run)?;
        let identify = |descriptor: &OwnedFd| {
            fs::fstat(descriptor).map_err(|source| {
                DispatchStoreError::io(
                    "identify locked dispatch run",
                    store.runs_root.join(run.as_str()),
                    source,
                )
            })
        };
        let expected = identify(&canonical_run)?;
        let actual = identify(access.run_fd)?;
        if expected.st_dev != actual.st_dev || expected.st_ino != actual.st_ino {
            return Err(DispatchStoreError::UnsafePath {
                path: store.runs_root.join(run.as_str()),
            });
        }
        let dispatch_fd = open_dispatch_dir(store, access.run_fd, run, false)?;
        operation(&LockedDispatchRun {
            store,
            run,
            dispatch: &dispatch_fd,
        })
    }

    pub(super) fn read_locked_inventory(
        authority: &LockedDispatchRun<'_>,
    ) -> DispatchStoreResult<DispatchInventory> {
        let path = authority
            .store
            .runs_root
            .join(authority.run.as_str())
            .join("dispatch");
        // A fresh open description avoids sharing a consumed directory offset
        // when validation reads this lock-owned inventory more than once.
        let descriptor = openat(
            authority.dispatch,
            ".",
            OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::empty(),
        )
        .map_err(|source| {
            DispatchStoreError::io("open dispatch inventory", path.clone(), source)
        })?;
        let mut directory = Dir::new(descriptor).map_err(|source| {
            DispatchStoreError::io("read dispatch inventory", path.clone(), source)
        })?;
        let mut inventory = DispatchInventory::default();
        for entry in &mut directory {
            let entry = entry.map_err(|source| {
                DispatchStoreError::io("read dispatch inventory", path.clone(), source)
            })?;
            let name = entry
                .file_name()
                .to_str()
                .map_err(|_| DispatchStoreError::UnsafePath {
                    path: path.join("<non-utf8>"),
                })?;
            match inventory_entry(name)? {
                Some(InventoryEntry::Record(agent)) => inventory.records.push(read_record(
                    authority.store,
                    authority.dispatch,
                    authority.run,
                    &agent,
                )?),
                Some(InventoryEntry::Pending(hash)) => inventory.pending.push(read_pending(
                    authority.store,
                    authority.dispatch,
                    authority.run,
                    hash,
                )?),
                None => {}
            }
        }
        sort_inventory(&mut inventory);
        Ok(inventory)
    }

    pub(super) fn load_pending(
        store: &DispatchStore,
        run: &RunId,
        launch_id_hash: [u8; 32],
    ) -> DispatchStoreResult<PendingDispatch> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, run, false).map_err(|error| {
            if is_not_found(&error) {
                DispatchStoreError::PendingNotFound { run: run.clone() }
            } else {
                error
            }
        })?;
        let _lock = acquire_lock(store, &run_fd, run)?;
        read_pending(store, &dispatch_fd, run, launch_id_hash)
    }

    pub(super) fn load_pending_for_agent(
        store: &DispatchStore,
        run: &RunId,
        agent_id: &AgentId,
    ) -> DispatchStoreResult<PendingDispatch> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, run, false)?;
        let _lock = acquire_lock(store, &run_fd, run)?;
        let path = store.runs_root.join(run.as_str()).join("dispatch");
        let duplicate = rustix::io::dup(&dispatch_fd).map_err(|source| {
            DispatchStoreError::io("duplicate dispatch directory", path.clone(), source)
        })?;
        let mut directory = Dir::new(duplicate).map_err(|source| {
            DispatchStoreError::io("read pending launches", path.clone(), source)
        })?;
        let mut matched = None;
        for entry in &mut directory {
            let entry = entry.map_err(|source| {
                DispatchStoreError::io("read pending launches", path.clone(), source)
            })?;
            let name = entry
                .file_name()
                .to_str()
                .map_err(|_| DispatchStoreError::UnsafePath {
                    path: path.join("<non-utf8>"),
                })?;
            let Some(hash) = name
                .strip_prefix("pending-")
                .and_then(|value| value.strip_suffix(".json"))
                .and_then(|value| parse_hash_name(value).ok())
            else {
                continue;
            };
            let pending = read_pending(store, &dispatch_fd, run, hash)?;
            if pending.expected_attachment.agent_id == *agent_id
                && matched.replace(pending).is_some()
            {
                return Err(DispatchStoreError::Domain(DispatchError::InvalidPending(
                    "multiple pending claims name one subject agent".into(),
                )));
            }
        }
        matched.ok_or_else(|| unknown_record(run, agent_id, "pending claim is absent"))
    }

    pub(super) fn claim_pending_unspawned<F>(
        store: &DispatchStore,
        run: &RunId,
        launch_id_hash: [u8; 32],
        child_process_hash: [u8; 32],
        validate: F,
    ) -> DispatchStoreResult<PendingDispatch>
    where
        F: FnOnce(&PendingDispatch, i64, &LockedDispatchRun<'_>) -> DispatchStoreResult<()>,
    {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, run, false)?;
        let _lock = acquire_lock(store, &run_fd, run)?;
        let now = current_time_millis();
        let state = store.load_run(run)?;
        if matches!(
            state.status.known(),
            Some(RunStatus::Closed | RunStatus::Closing)
        ) {
            return Err(DispatchStoreError::PendingPath { run: run.clone() });
        }
        let pending = read_pending(store, &dispatch_fd, run, launch_id_hash)?;
        if state.status != pending.run_status {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidPending(
                "run state changed during pending claim".into(),
            )));
        }
        let authority = LockedDispatchRun {
            store,
            run,
            dispatch: &dispatch_fd,
        };
        validate(&pending, now, &authority)?;
        pending.check_lease(now)?;
        let mut claimed = pending;
        claimed.claim(now, child_process_hash)?;
        replace_existing(
            store,
            &dispatch_fd,
            run,
            &pending_name(launch_id_hash),
            &encode_document(&claimed)?,
        )?;
        Ok(claimed)
    }

    pub(super) fn activate_pending<F>(
        store: &DispatchStore,
        run: &RunId,
        launch_id_hash: [u8; 32],
        child_process_hash: [u8; 32],
        activate: F,
    ) -> DispatchStoreResult<DispatchRecord>
    where
        F: FnOnce(
            &PendingDispatch,
            i64,
            &LockedDispatchRun<'_>,
        ) -> DispatchStoreResult<DispatchRecord>,
    {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, run, false)?;
        let _lock = acquire_lock(store, &run_fd, run)?;
        let now = current_time_millis();
        let state = store.load_run(run)?;
        if matches!(
            state.status.known(),
            Some(RunStatus::Closed | RunStatus::Closing)
        ) {
            return Err(DispatchStoreError::PendingPath { run: run.clone() });
        }
        let pending = read_pending(store, &dispatch_fd, run, launch_id_hash)?;
        if state.status != pending.run_status {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidPending(
                "run state changed during pending activation".into(),
            )));
        }
        if pending.launch_state != shepherd::dispatch::PendingLaunchState::ClaimedUnspawned {
            return Err(DispatchStoreError::Domain(
                DispatchError::PendingLaunchConsumed,
            ));
        }
        if pending.child_process_hash.as_ref() != Some(&child_process_hash) {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidPending(
                "activation process identity does not match the claimed child".into(),
            )));
        }
        let authority = LockedDispatchRun {
            store,
            run,
            dispatch: &dispatch_fd,
        };
        let record = activate(&pending, now, &authority)?;
        let mut active = pending;
        active.activate(now, child_process_hash)?;
        publish_no_clobber(
            store,
            &dispatch_fd,
            run,
            &record_name(&record.agent_id),
            &encode_record(&record)?,
        )?;
        replace_existing(
            store,
            &dispatch_fd,
            run,
            &pending_name(launch_id_hash),
            &encode_document(&active)?,
        )?;
        Ok(record)
    }

    pub(super) fn reconcile_unspawned(
        store: &DispatchStore,
        run: &RunId,
        requested_now: Option<i64>,
    ) -> DispatchStoreResult<usize> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, run, false)?;
        let _lock = acquire_lock(store, &run_fd, run)?;
        let now = requested_now.unwrap_or_else(current_time_millis);
        let dispatch_path = store.runs_root.join(run.as_str()).join("dispatch");
        let duplicate = rustix::io::dup(&dispatch_fd).map_err(|source| {
            DispatchStoreError::io(
                "duplicate dispatch directory",
                dispatch_path.clone(),
                source,
            )
        })?;
        let mut directory = Dir::new(duplicate).map_err(|source| {
            DispatchStoreError::io("read pending launches", dispatch_path.clone(), source)
        })?;
        let mut expired = 0;
        for entry in &mut directory {
            let entry = entry.map_err(|source| {
                DispatchStoreError::io("read pending launches", dispatch_path.clone(), source)
            })?;
            let name = entry
                .file_name()
                .to_str()
                .map_err(|_| DispatchStoreError::UnsafePath {
                    path: dispatch_path.join("<non-utf8>"),
                })?
                .to_owned();
            if !name.starts_with("pending-") || !name.ends_with(".json") {
                continue;
            }
            let Some(hex_hash) = name
                .strip_prefix("pending-")
                .and_then(|value| value.strip_suffix(".json"))
            else {
                continue;
            };
            let Ok(launch_hash) = parse_hash_name(hex_hash) else {
                continue;
            };
            let mut pending = read_pending(store, &dispatch_fd, run, launch_hash)?;
            let agent_id = pending.expected_attachment.agent_id.clone();
            let active_record = match read_record(store, &dispatch_fd, run, &agent_id) {
                Ok(record) if record.state == DispatchState::Active => Some(record),
                Ok(_) => None,
                Err(DispatchStoreError::UnknownRecord { reason, .. })
                    if reason == "record is absent" =>
                {
                    None
                }
                Err(error) => return Err(error),
            };
            let should_reconcile = match pending.launch_state {
                PendingLaunchState::ClaimedUnspawned => true,
                PendingLaunchState::Pending => now >= pending.expires_at,
                PendingLaunchState::Active => active_record.is_none(),
                PendingLaunchState::Quarantined
                | PendingLaunchState::LaunchFailed
                | PendingLaunchState::Canceled
                | PendingLaunchState::Expired => false,
            };
            if should_reconcile {
                if pending.launch_state == PendingLaunchState::ClaimedUnspawned
                    || (pending.launch_state == PendingLaunchState::Pending
                        && now >= pending.expires_at)
                {
                    if active_record.is_some() {
                        remove_record(store, &dispatch_fd, run, &agent_id)?;
                    }
                    pending.expire()?;
                } else {
                    pending.fail_after_recovery()?;
                }
                replace_existing(
                    store,
                    &dispatch_fd,
                    run,
                    &pending_name(launch_hash),
                    &encode_document(&pending)?,
                )?;
                expired += 1;
            }
        }
        Ok(expired)
    }

    pub(super) fn cleanup_pending(
        store: &DispatchStore,
        run: &RunId,
        launch_hash: [u8; 32],
        state: PendingLaunchState,
    ) -> DispatchStoreResult<LaunchCleanupResponse> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, run, false)?;
        let _lock = acquire_lock(store, &run_fd, run)?;
        let mut pending = read_pending(store, &dispatch_fd, run, launch_hash)?;
        if pending.launch_state.is_terminal() {
            return Err(DispatchStoreError::Domain(
                DispatchError::PendingLaunchConsumed,
            ));
        }
        match state {
            PendingLaunchState::Expired => pending.expire()?,
            PendingLaunchState::Canceled => pending.cancel()?,
            PendingLaunchState::LaunchFailed => pending.fail()?,
            PendingLaunchState::Quarantined => {
                return Err(DispatchStoreError::Domain(DispatchError::InvalidPending(
                    "malignant quarantine is not a caller-selected cleanup state".into(),
                )));
            }
            PendingLaunchState::Pending
            | PendingLaunchState::ClaimedUnspawned
            | PendingLaunchState::Active => {
                return Err(DispatchStoreError::Domain(DispatchError::InvalidPending(
                    "launch cleanup target is not terminal".into(),
                )));
            }
        }
        replace_existing(
            store,
            &dispatch_fd,
            run,
            &pending_name(launch_hash),
            &encode_document(&pending)?,
        )?;
        let response = LaunchCleanupResponse {
            schema: shepherd::dispatch::LAUNCH_CLEANUP_SCHEMA.into(),
            launch_id_hash: launch_hash,
            state,
        };
        response.validate()?;
        Ok(response)
    }

    pub(super) fn load_root_binding(
        store: &DispatchStore,
        run: &RunId,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, run, false)?;
        let _lock = acquire_lock(store, &run_fd, run)?;
        read_root_binding_at(store, &dispatch_fd, run, session_id)
    }

    pub(super) fn read_locked_root_binding(
        authority: &LockedDispatchRun<'_>,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        read_root_binding_at(
            authority.store,
            authority.dispatch,
            authority.run,
            session_id,
        )
    }

    pub(super) fn read_locked_review_custody(
        authority: &LockedDispatchRun<'_>,
        subject: &AgentId,
    ) -> DispatchStoreResult<ReviewCustody> {
        // A terminal result denies the transition. Recovery remains the job of
        // the normal custody/record loader, never a nested transition here.
        read_review_custody_at(authority.store, authority.dispatch, authority.run, subject)
    }

    pub(super) fn read_locked_record(
        authority: &LockedDispatchRun<'_>,
        subject: &AgentId,
    ) -> DispatchStoreResult<DispatchRecord> {
        read_record(authority.store, authority.dispatch, authority.run, subject)
    }

    pub(super) fn read_locked_pending(
        authority: &LockedDispatchRun<'_>,
        launch_hash: [u8; 32],
    ) -> DispatchStoreResult<PendingDispatch> {
        read_pending(
            authority.store,
            authority.dispatch,
            authority.run,
            launch_hash,
        )
    }

    pub(super) fn load_profile_lease(
        store: &DispatchStore,
        run: &RunId,
        session_id: &SessionId,
    ) -> DispatchStoreResult<ProfileLease> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, run, false)?;
        let _lock = acquire_lock(store, &run_fd, run)?;
        read_profile_lease_at(store, &dispatch_fd, run, session_id)
    }

    fn read_profile_lease_at(
        store: &DispatchStore,
        dispatch_fd: &OwnedFd,
        run: &RunId,
        session_id: &SessionId,
    ) -> DispatchStoreResult<ProfileLease> {
        let name = profile_lease_name(session_id);
        let path = store.profile_lease_path(run, session_id);
        let file = open_regular_at(dispatch_fd, &name, &path)?;
        let bytes = read_bounded(file, MAX_RECORD_BYTES)
            .map_err(|source| DispatchStoreError::io("read profile lease", path, source))?;
        decode_profile_lease(&bytes, run, session_id)
    }

    pub(super) fn load_review_custody(
        store: &DispatchStore,
        run: &RunId,
        subject: &AgentId,
    ) -> DispatchStoreResult<ReviewCustody> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, run, false)?;
        let _lock = acquire_lock(store, &run_fd, run)?;
        let custody = read_review_custody_at(store, &dispatch_fd, run, subject)?;
        if custody.state != ReviewCustodyState::Active {
            recover_malignant_subject(store, &dispatch_fd, &custody, false)?;
        }
        Ok(custody)
    }

    fn read_review_custody_at(
        store: &DispatchStore,
        dispatch_fd: &OwnedFd,
        run: &RunId,
        subject: &AgentId,
    ) -> DispatchStoreResult<ReviewCustody> {
        let name = review_custody_name(subject);
        let path = store.review_custody_path(run, subject);
        let file = open_regular_at(dispatch_fd, &name, &path)?;
        let bytes = read_bounded(file, MAX_RECORD_BYTES)
            .map_err(|source| DispatchStoreError::io("read review custody", path, source))?;
        decode_review_custody(&bytes, run, subject)
    }

    pub(super) fn read_review_terminal_snapshot(
        store: &DispatchStore,
        run: &RunId,
        subject: &AgentId,
    ) -> DispatchStoreResult<ReviewTerminalSnapshot> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, run, false)?;
        let _lock = acquire_lock(store, &run_fd, run)?;
        let custody = read_review_custody_at(store, &dispatch_fd, run, subject)?;
        Ok(ReviewTerminalSnapshot {
            root: read_root_binding_at(store, &dispatch_fd, run, &custody.root_session_id)?,
            subject: read_record(store, &dispatch_fd, run, subject)?,
            pending: read_pending(store, &dispatch_fd, run, custody.pending_launch_id_hash)?,
            custody,
        })
    }

    pub(super) fn load_skill_use(
        store: &DispatchStore,
        run: &RunId,
        key: SkillUseKey<'_>,
        skill: &str,
    ) -> DispatchStoreResult<SkillUseChallenge> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, run, false)?;
        let _lock = acquire_lock(store, &run_fd, run)?;
        validate_skill_use_root_at(store, &root, &dispatch_fd, run, key)?;
        read_skill_use_at(store, &dispatch_fd, run, key, skill)
    }

    fn validate_skill_use_root_at(
        store: &DispatchStore,
        runs_fd: &OwnedFd,
        dispatch_fd: &OwnedFd,
        run: &RunId,
        key: SkillUseKey<'_>,
    ) -> DispatchStoreResult<()> {
        let SkillUseKey::Root(expected) = key else {
            return Ok(());
        };
        let session = &expected.binding.session_id;
        let binding = read_root_binding_at(store, dispatch_fd, run, session)?;
        let current_path = store.current_root_binding_path(session);
        let current = open_regular_at(runs_fd, &root_binding_name(session), &current_path)?;
        let current = read_bounded(current, MAX_RECORD_BYTES).map_err(|source| {
            DispatchStoreError::io("read current root binding", current_path, source)
        })?;
        let current = decode_current_root_binding(&current, session)?;
        let profile = match read_profile_lease_at(store, dispatch_fd, run, session) {
            Ok(lease) => Some(lease),
            Err(DispatchStoreError::Io { source, .. })
                if source.kind() == std::io::ErrorKind::NotFound =>
            {
                None
            }
            Err(error) => return Err(error),
        };
        if binding != expected.binding || current != binding || profile != expected.profile_lease {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidSkillUse(
                "root or profile authority changed during the skill-use operation".into(),
            )));
        }
        Ok(())
    }

    fn read_skill_use_at(
        store: &DispatchStore,
        dispatch_fd: &OwnedFd,
        run: &RunId,
        key: SkillUseKey<'_>,
        skill: &str,
    ) -> DispatchStoreResult<SkillUseChallenge> {
        let name = skill_use_name(key, skill)?;
        let path = store
            .runs_root
            .join(run.as_str())
            .join("dispatch")
            .join(&name);
        let file = open_regular_at(dispatch_fd, &name, &path)?;
        let bytes = read_bounded(file, MAX_RECORD_BYTES)
            .map_err(|source| DispatchStoreError::io("read skill-use challenge", path, source))?;
        decode_skill_use(&bytes, run, key, skill)
    }

    fn read_root_binding_at(
        store: &DispatchStore,
        dispatch_fd: &OwnedFd,
        run: &RunId,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        let root = open_runs_root(store)?;
        ensure_no_root_renewal_at(&root, session_id)?;
        read_root_binding_raw_at(store, dispatch_fd, run, session_id)
    }

    fn read_root_binding_raw_at(
        store: &DispatchStore,
        dispatch_fd: &OwnedFd,
        run: &RunId,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        let name = root_binding_name(session_id);
        let path = store.root_binding_path(run, session_id);
        let file = open_regular_at(dispatch_fd, &name, &path)?;
        let bytes = read_bounded(file, MAX_RECORD_BYTES)
            .map_err(|source| DispatchStoreError::io("read root binding", path.clone(), source))?;
        decode_root_binding(&bytes, run, session_id)
    }

    pub(super) fn load_run(store: &DispatchStore, run: &RunId) -> DispatchStoreResult<RunState> {
        let root = open_runs_root(store)?;
        read_run_document(store, &root, run)
    }

    pub(super) fn load(
        store: &DispatchStore,
        run: &RunId,
        agent_id: &AgentId,
    ) -> DispatchStoreResult<DispatchRecord> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, run, false).map_err(|error| {
            if is_not_found(&error) {
                unknown_record(run, agent_id, "dispatch directory is absent")
            } else {
                error
            }
        })?;
        let _lock = acquire_lock(store, &run_fd, run)?;
        let record = read_record(store, &dispatch_fd, run, agent_id)?;
        match read_review_custody_at(store, &dispatch_fd, run, agent_id) {
            Ok(custody) if custody.state != ReviewCustodyState::Active => {
                recover_malignant_subject(store, &dispatch_fd, &custody, false)
                    .map(|(record, _)| record)
            }
            Ok(_) => Ok(record),
            Err(error) if is_not_found(&error) => Ok(record),
            Err(error) => Err(error),
        }
    }

    pub(super) fn read_artifact(
        store: &DispatchStore,
        run: &RunId,
        reference: &str,
    ) -> DispatchStoreResult<Vec<u8>> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let mut directory = run_fd;
        let parts = reference.split('/').collect::<Vec<_>>();
        for part in &parts[..parts.len() - 1] {
            directory = openat(
                &directory,
                *part,
                OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
                Mode::empty(),
            )
            .map_err(|source| {
                unsafe_path_or_io(
                    "open artifact directory",
                    store.runs_root.join(run.as_str()).join(reference),
                    source,
                )
            })?;
        }
        let path = store.runs_root.join(run.as_str()).join(reference);
        let file = open_regular_at(&directory, parts[parts.len() - 1], &path)?;
        read_bounded(file, MAX_RECORD_BYTES)
            .map_err(|source| DispatchStoreError::io("read completion artifact", path, source))
    }

    pub(super) fn stop(
        store: &DispatchStore,
        run: &RunId,
        request: StopRequest,
    ) -> DispatchStoreResult<DispatchRecord> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, run, false)?;
        let _lock = acquire_lock(store, &run_fd, run)?;
        let mut record = read_record(store, &dispatch_fd, run, &request.agent_id)?;
        let mut request = request;
        request.stopped_at = current_time_millis();
        record.stop(request)?;
        record.validate_loaded()?;
        replace_record(store, &dispatch_fd, run, &record)?;
        Ok(record)
    }

    pub(super) fn stop_verified(
        store: &DispatchStore,
        run: &RunId,
        native: &NativeIdentity,
        request: StopRequest,
    ) -> DispatchStoreResult<DispatchRecord> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, run, false)?;
        let _lock = acquire_lock(store, &run_fd, run)?;
        let mut record = read_record(store, &dispatch_fd, run, &request.agent_id)?;
        resolve_native_identity(Some(&record), native)?;
        let mut request = request;
        request.stopped_at = current_time_millis();
        record.stop(request)?;
        record.validate_loaded()?;
        replace_record(store, &dispatch_fd, run, &record)?;
        Ok(record)
    }

    pub(super) fn quarantine_malignant(
        store: &DispatchStore,
        expected_custody: Option<&ReviewCustody>,
        expected: &DispatchRecord,
        expected_pending: &PendingDispatch,
        custody: &ReviewCustody,
    ) -> DispatchStoreResult<(DispatchRecord, PendingDispatch)> {
        let root = open_runs_root(store)?;
        let run_fd = open_run_dir(store, &root, &expected.run)?;
        let dispatch_fd = open_dispatch_dir(store, &run_fd, &expected.run, false)?;
        let _lock = acquire_lock(store, &run_fd, &expected.run)?;
        let current_custody =
            read_review_custody_at(store, &dispatch_fd, &expected.run, &expected.agent_id);
        let already_committed = current_custody
            .as_ref()
            .is_ok_and(|current| current == custody);
        if !already_committed {
            match (expected_custody, current_custody) {
                (Some(expected_custody), Ok(current)) if &current == expected_custody => {
                    replace_existing(
                        store,
                        &dispatch_fd,
                        &expected.run,
                        &review_custody_name(&expected.agent_id),
                        &encode_document(custody)?,
                    )?;
                }
                (None, Err(error)) if is_not_found(&error) => {
                    publish_no_clobber(
                        store,
                        &dispatch_fd,
                        &expected.run,
                        &review_custody_name(&expected.agent_id),
                        &encode_document(custody)?,
                    )?;
                }
                _ => {
                    return Err(DispatchStoreError::Domain(
                        DispatchError::InvalidReviewCustody(
                            "review custody changed before subject quarantine".into(),
                        ),
                    ));
                }
            }
            if store.review_quarantine_fault == Some(ReviewQuarantineFault::AfterCustodyCommit) {
                return Err(DispatchStoreError::Reconciliation(
                    "injected failure after malignant custody commit".into(),
                ));
            }
        }
        let current_record = read_record(store, &dispatch_fd, &expected.run, &expected.agent_id)?;
        let current_pending = read_pending(
            store,
            &dispatch_fd,
            &expected.run,
            expected_pending.launch_id_hash,
        )?;
        if current_record != *expected && current_record.state != DispatchState::Malignant {
            return Err(DispatchStoreError::Domain(
                DispatchError::InvalidReviewCustody(
                    "subject dispatch changed before malignant quarantine".into(),
                ),
            ));
        }
        if current_pending != *expected_pending
            && current_pending.launch_state != PendingLaunchState::Quarantined
        {
            return Err(DispatchStoreError::Domain(
                DispatchError::InvalidReviewCustody(
                    "subject pending claim changed before malignant quarantine".into(),
                ),
            ));
        }
        recover_malignant_subject(store, &dispatch_fd, custody, !already_committed)
    }

    fn recover_malignant_subject(
        store: &DispatchStore,
        dispatch_fd: &OwnedFd,
        custody: &ReviewCustody,
        inject_fault: bool,
    ) -> DispatchStoreResult<(DispatchRecord, PendingDispatch)> {
        let stopped_at = custody.stopped_at.ok_or_else(|| {
            DispatchStoreError::Domain(DispatchError::InvalidReviewCustody(
                "terminal review custody has no stop time".into(),
            ))
        })?;
        let mut record = read_record(store, dispatch_fd, &custody.run, &custody.subject_agent_id)?;
        if record.root_session_id != custody.root_session_id
            || record.session_id != custody.subject_session_id
            || record.role != custody.subject_role
            || record.lane != custody.lane
        {
            return Err(DispatchStoreError::Domain(
                DispatchError::InvalidReviewCustody(
                    "terminal custody does not match the durable dispatch identity".into(),
                ),
            ));
        }
        match record.state {
            DispatchState::Active => {
                record.quarantine_malignant(stopped_at)?;
                replace_record(store, dispatch_fd, &custody.run, &record)?;
            }
            DispatchState::Malignant if record.stopped_at == Some(stopped_at) => {}
            _ => {
                return Err(DispatchStoreError::Domain(
                    DispatchError::InvalidReviewCustody(
                        "terminal custody conflicts with the durable dispatch state".into(),
                    ),
                ));
            }
        }
        if inject_fault
            && store.review_quarantine_fault == Some(ReviewQuarantineFault::AfterRecordCommit)
        {
            return Err(DispatchStoreError::Reconciliation(
                "injected failure after malignant dispatch commit".into(),
            ));
        }
        let mut pending = read_pending(
            store,
            dispatch_fd,
            &custody.run,
            custody.pending_launch_id_hash,
        )?;
        if pending.root_session_id != custody.root_session_id
            || pending.expected_attachment.agent_id != custody.subject_agent_id
            || pending.expected_child_session_id != custody.subject_session_id
        {
            return Err(DispatchStoreError::Domain(
                DispatchError::InvalidReviewCustody(
                    "terminal custody does not match the durable pending claim".into(),
                ),
            ));
        }
        match pending.launch_state {
            PendingLaunchState::Active => {
                pending.quarantine()?;
                replace_existing(
                    store,
                    dispatch_fd,
                    &custody.run,
                    &pending_name(pending.launch_id_hash),
                    &encode_document(&pending)?,
                )?;
            }
            PendingLaunchState::Quarantined => {}
            _ => {
                return Err(DispatchStoreError::Domain(
                    DispatchError::InvalidReviewCustody(
                        "terminal custody conflicts with the durable pending claim".into(),
                    ),
                ));
            }
        }
        Ok((record, pending))
    }

    /// Walk `/` down to the runs root one `O_NOFOLLOW` descriptor at a time.
    ///
    /// Also the anchor the shared resolver holds across enumeration, which is
    /// why an unsafe runs root is `UnsafePath` before a single entry is read.
    pub(super) fn open_runs_root(store: &DispatchStore) -> DispatchStoreResult<OwnedFd> {
        reject_parent_components(&store.runs_root)?;
        let mut descriptor = open(
            "/",
            OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::empty(),
        )
        .map_err(|source| unsafe_path_or_io("open filesystem root", PathBuf::from("/"), source))?;
        let mut traversed = PathBuf::from("/");
        for component in store.runs_root.components() {
            let std::path::Component::Normal(name) = component else {
                continue;
            };
            traversed.push(name);
            descriptor = openat(
                &descriptor,
                name,
                OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
                Mode::empty(),
            )
            .map_err(|source| {
                unsafe_path_or_io("open runs root component", traversed.clone(), source)
            })?;
        }
        Ok(descriptor)
    }

    pub(super) fn run_names(
        store: &DispatchStore,
        root: &OwnedFd,
    ) -> DispatchStoreResult<Vec<String>> {
        let duplicate = rustix::io::dup(root).map_err(|source| {
            DispatchStoreError::io("duplicate runs root", store.runs_root.clone(), source)
        })?;
        let mut directory = Dir::new(duplicate).map_err(|source| {
            DispatchStoreError::io("read runs directory", store.runs_root.clone(), source)
        })?;
        let mut names = Vec::new();
        for entry in &mut directory {
            let entry = entry.map_err(|source| {
                DispatchStoreError::io("read runs directory", store.runs_root.clone(), source)
            })?;
            let name = entry
                .file_name()
                .to_str()
                .map_err(|_| DispatchStoreError::UnsafePath {
                    path: store.runs_root.join("<non-utf8>"),
                })?;
            names.push(name.to_owned());
        }
        Ok(names)
    }

    fn open_run_dir(
        store: &DispatchStore,
        root: &OwnedFd,
        run: &RunId,
    ) -> DispatchStoreResult<OwnedFd> {
        let path = store.runs_root.join(run.as_str());
        openat(
            root,
            run.as_str(),
            OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::empty(),
        )
        .map_err(|source| unsafe_path_or_io("open run directory", path, source))
    }

    fn open_dispatch_dir(
        store: &DispatchStore,
        run_fd: &OwnedFd,
        run: &RunId,
        create: bool,
    ) -> DispatchStoreResult<OwnedFd> {
        let path = store.runs_root.join(run.as_str()).join("dispatch");
        let flags = OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW;
        match openat(run_fd, "dispatch", flags, Mode::empty()) {
            Ok(fd) => Ok(fd),
            Err(source) if create && source == rustix::io::Errno::NOENT => {
                match mkdirat(run_fd, "dispatch", Mode::RWXU) {
                    Ok(()) | Err(rustix::io::Errno::EXIST) => {}
                    Err(source) => {
                        return Err(DispatchStoreError::io(
                            "create dispatch directory",
                            path,
                            source,
                        ));
                    }
                }
                openat(run_fd, "dispatch", flags, Mode::empty())
                    .map_err(|source| unsafe_path_or_io("open dispatch directory", path, source))
            }
            Err(source) => Err(unsafe_path_or_io("open dispatch directory", path, source)),
        }
    }

    /// Read `<runs_root>/<run>/run.json` through the runs-root descriptor.
    ///
    /// Opening the run directory is part of the read, not a separate step the
    /// caller sequences: a run whose directory vanished between enumeration
    /// and here is *absent*, which is the one failure the shared resolver
    /// skips. A linked run directory is `ELOOP` and a non-directory is
    /// `ENOTDIR`; `unsafe_path_or_io` renders both as `UnsafePath`, so an
    /// escape attempt can never be mistaken for absence.
    pub(super) fn read_run_document(
        store: &DispatchStore,
        root: &OwnedFd,
        run: &RunId,
    ) -> DispatchStoreResult<RunState> {
        let run_fd = open_run_dir(store, root, run)?;
        let path = store.runs_root.join(run.as_str()).join("run.json");
        let file = open_regular_at(&run_fd, "run.json", &path)?;
        let bytes = read_bounded(file, MAX_RECORD_BYTES)
            .map_err(|source| DispatchStoreError::io("read run document", path.clone(), source))?;
        let state: RunState = serde_json::from_slice(&bytes).map_err(|error| {
            DispatchStoreError::InvalidRunDocument {
                path: path.clone(),
                reason: error.to_string(),
            }
        })?;
        if state.run != run.as_str()
            || state.schema_version != 1
            || !matches!(
                state.status.as_str(),
                "planted" | "planned" | "executing" | "closing" | "closed"
            )
        {
            return Err(DispatchStoreError::InvalidRunDocument {
                path,
                reason: "run identity, schema, or status is invalid".into(),
            });
        }
        Ok(state)
    }

    fn read_pending(
        store: &DispatchStore,
        dispatch_fd: &OwnedFd,
        run: &RunId,
        launch_id_hash: [u8; 32],
    ) -> DispatchStoreResult<PendingDispatch> {
        let name = pending_name(launch_id_hash);
        let path = store
            .runs_root
            .join(run.as_str())
            .join("dispatch")
            .join(&name);
        let file = match open_regular_at(dispatch_fd, &name, &path) {
            Ok(file) => file,
            Err(DispatchStoreError::Io { source, .. })
                if source.kind() == std::io::ErrorKind::NotFound =>
            {
                return Err(DispatchStoreError::PendingNotFound { run: run.clone() });
            }
            Err(error) => return Err(error),
        };
        let bytes = read_bounded(file, MAX_RECORD_BYTES).map_err(|source| {
            DispatchStoreError::io("read pending dispatch", path.clone(), source)
        })?;
        let pending: PendingDispatch = serde_json::from_slice(&bytes).map_err(|error| {
            DispatchStoreError::Domain(DispatchError::InvalidRecord(error.to_string()))
        })?;
        pending.validate()?;
        if pending.run != *run || !constant_time_digest_eq(&pending.launch_id_hash, &launch_id_hash)
        {
            return Err(DispatchStoreError::PendingPath { run: run.clone() });
        }
        Ok(pending)
    }

    fn read_record(
        store: &DispatchStore,
        dispatch_fd: &OwnedFd,
        run: &RunId,
        agent_id: &AgentId,
    ) -> DispatchStoreResult<DispatchRecord> {
        let name = record_name(agent_id);
        let path = store.record_path(run, agent_id);
        let file = match open_regular_at(dispatch_fd, &name, &path) {
            Ok(file) => file,
            Err(DispatchStoreError::Io { source, .. })
                if source.kind() == std::io::ErrorKind::NotFound =>
            {
                return Err(unknown_record(run, agent_id, "record is absent"));
            }
            Err(error) => return Err(error),
        };
        let bytes = read_bounded(file, MAX_RECORD_BYTES)
            .map_err(|error| unknown_record(run, agent_id, error.to_string()))?;
        let record: DispatchRecord = serde_json::from_slice(&bytes)
            .map_err(|error| unknown_record(run, agent_id, error.to_string()))?;
        record
            .validate_loaded()
            .map_err(|error| unknown_record(run, agent_id, error.to_string()))?;
        if &record.agent_id != agent_id || &record.run != run {
            return Err(unknown_record(
                run,
                agent_id,
                "record identity does not match its canonical path",
            ));
        }
        Ok(record)
    }

    fn open_regular_at(parent: &OwnedFd, name: &str, path: &Path) -> DispatchStoreResult<File> {
        let fd = openat(
            parent,
            name,
            OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::empty(),
        )
        .map_err(|source| unsafe_path_or_io("open regular file", path.to_path_buf(), source))?;
        let stat = fs::fstat(&fd).map_err(|source| {
            DispatchStoreError::io("inspect regular file", path.to_path_buf(), source)
        })?;
        if !FileType::from_raw_mode(stat.st_mode).is_file() || stat.st_nlink != 1 {
            return Err(DispatchStoreError::UnsafePath {
                path: path.to_path_buf(),
            });
        }
        Ok(File::from(fd))
    }

    fn remove_record(
        store: &DispatchStore,
        dispatch_fd: &OwnedFd,
        run: &RunId,
        agent_id: &AgentId,
    ) -> DispatchStoreResult<()> {
        let name = record_name(agent_id);
        let path = store.record_path(run, agent_id);
        let stat = match fs::statat(dispatch_fd, &name, AtFlags::SYMLINK_NOFOLLOW) {
            Ok(stat) => stat,
            Err(rustix::io::Errno::NOENT) => return Ok(()),
            Err(error) => return Err(unsafe_path_or_io("inspect active record", path, error)),
        };
        if !FileType::from_raw_mode(stat.st_mode).is_file() || stat.st_nlink != 1 {
            return Err(DispatchStoreError::UnsafePath { path });
        }
        unlinkat(dispatch_fd, &name, AtFlags::empty()).map_err(|error| {
            unsafe_path_or_io("remove orphaned active record", path.clone(), error)
        })?;
        fs::fsync(dispatch_fd)
            .map_err(|error| DispatchStoreError::io("fsync dispatch directory", path, error))
    }

    fn acquire_lock(
        store: &DispatchStore,
        run_fd: &OwnedFd,
        run: &RunId,
    ) -> DispatchStoreResult<DispatchLock> {
        let path = store.runs_root.join(run.as_str()).join("run.lock");
        let fd = openat(
            run_fd,
            "run.lock",
            OFlags::RDWR | OFlags::CREATE | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::RUSR | Mode::WUSR,
        )
        .map_err(|source| unsafe_path_or_io("open dispatch lock", path.clone(), source))?;
        let file = File::from(fd);
        let started = Instant::now();
        loop {
            match file.try_lock() {
                Ok(()) => return Ok(DispatchLock(file)),
                Err(TryLockError::WouldBlock) if started.elapsed() < store.timeout => {
                    let remaining = store.timeout.saturating_sub(started.elapsed());
                    std::thread::sleep(LOCK_RETRY_INTERVAL.min(remaining));
                }
                Err(TryLockError::WouldBlock) => {
                    return Err(DispatchStoreError::LockTimeout {
                        path,
                        timeout: store.timeout,
                    });
                }
                Err(TryLockError::Error(source)) => {
                    return Err(DispatchStoreError::io(
                        "acquire dispatch lock",
                        path,
                        source,
                    ));
                }
            }
        }
    }

    fn acquire_current_root_lock(
        store: &DispatchStore,
        root: &OwnedFd,
    ) -> DispatchStoreResult<DispatchLock> {
        let path = store.runs_root.join(".root-session.lock");
        let fd = openat(
            root,
            ".root-session.lock",
            OFlags::RDWR | OFlags::CREATE | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::RUSR | Mode::WUSR,
        )
        .map_err(|source| unsafe_path_or_io("open current root lock", path.clone(), source))?;
        let file = File::from(fd);
        let started = Instant::now();
        loop {
            match file.try_lock() {
                Ok(()) => return Ok(DispatchLock(file)),
                Err(TryLockError::WouldBlock) if started.elapsed() < store.timeout => {
                    let remaining = store.timeout.saturating_sub(started.elapsed());
                    std::thread::sleep(LOCK_RETRY_INTERVAL.min(remaining));
                }
                Err(TryLockError::WouldBlock) => {
                    return Err(DispatchStoreError::LockTimeout {
                        path,
                        timeout: store.timeout,
                    });
                }
                Err(TryLockError::Error(source)) => {
                    return Err(DispatchStoreError::io(
                        "acquire current root lock",
                        path,
                        source,
                    ));
                }
            }
        }
    }

    fn ensure_preparing_file(
        store: &DispatchStore,
        dispatch_fd: &OwnedFd,
        run: &RunId,
        name: &str,
        bytes: &[u8],
    ) -> DispatchStoreResult<()> {
        match openat(
            dispatch_fd,
            name,
            OFlags::WRONLY | OFlags::CREATE | OFlags::EXCL | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::RUSR | Mode::WUSR,
        ) {
            Ok(fd) => {
                let mut file = File::from(fd);
                file.write_all(bytes).map_err(|source| {
                    DispatchStoreError::io(
                        "write singleton preparing file",
                        store
                            .runs_root
                            .join(run.as_str())
                            .join("dispatch")
                            .join(name),
                        source,
                    )
                })?;
                file.sync_all().map_err(|source| {
                    DispatchStoreError::io(
                        "fsync singleton preparing file",
                        store
                            .runs_root
                            .join(run.as_str())
                            .join("dispatch")
                            .join(name),
                        source,
                    )
                })?;
                fs::fsync(dispatch_fd).map_err(|source| {
                    DispatchStoreError::io(
                        "fsync singleton dispatch directory",
                        store.runs_root.join(run.as_str()).join("dispatch"),
                        source,
                    )
                })
            }
            Err(source) if source == rustix::io::Errno::EXIST => {
                let existing = read_optional_at(dispatch_fd, name, MAX_RECORD_BYTES)?;
                if existing.as_deref() == Some(bytes) {
                    Ok(())
                } else {
                    Err(DispatchStoreError::SingletonPublication {
                        nonce: name.into(),
                        reason: "preparing file already contains different bytes".into(),
                    })
                }
            }
            Err(source) => Err(DispatchStoreError::io(
                "create singleton preparing file",
                store
                    .runs_root
                    .join(run.as_str())
                    .join("dispatch")
                    .join(name),
                source,
            )),
        }
    }

    fn publish_final_no_clobber(
        store: &DispatchStore,
        dispatch_fd: &OwnedFd,
        run: &RunId,
        temp: &str,
        target: &str,
        bytes: &[u8],
    ) -> DispatchStoreResult<()> {
        if let Some(existing) = read_optional_at(dispatch_fd, target, MAX_RECORD_BYTES)? {
            if existing != bytes {
                return Err(DispatchStoreError::SingletonPublication {
                    nonce: temp.into(),
                    reason: "final dispatch name contains different bytes".into(),
                });
            }
            remove_optional_at(dispatch_fd, temp)?;
        } else {
            match linkat(dispatch_fd, temp, dispatch_fd, target, AtFlags::empty()) {
                Ok(()) => {
                    remove_optional_at(dispatch_fd, temp)?;
                }
                Err(source) if source == rustix::io::Errno::EXIST => {
                    let existing = read_optional_at(dispatch_fd, target, MAX_RECORD_BYTES)?;
                    if existing.as_deref() != Some(bytes) {
                        return Err(DispatchStoreError::SingletonPublication {
                            nonce: temp.into(),
                            reason: "racing final dispatch publication differs".into(),
                        });
                    }
                    remove_optional_at(dispatch_fd, temp)?;
                }
                Err(source) => {
                    return Err(DispatchStoreError::io(
                        "publish singleton dispatch record",
                        store
                            .runs_root
                            .join(run.as_str())
                            .join("dispatch")
                            .join(target),
                        source,
                    ));
                }
            }
        }
        fs::fsync(dispatch_fd).map_err(|source| {
            DispatchStoreError::io(
                "fsync singleton dispatch directory",
                store.runs_root.join(run.as_str()).join("dispatch"),
                source,
            )
        })
    }

    fn read_optional_at(
        parent: &OwnedFd,
        name: &str,
        limit: u64,
    ) -> DispatchStoreResult<Option<Vec<u8>>> {
        match open_regular_at(parent, name, Path::new(name)) {
            Ok(file) => read_bounded(file, limit).map(Some).map_err(|source| {
                DispatchStoreError::io("read singleton publication", name.into(), source)
            }),
            Err(DispatchStoreError::Io { source, .. })
                if source.kind() == std::io::ErrorKind::NotFound =>
            {
                Ok(None)
            }
            Err(error) => Err(error),
        }
    }

    fn remove_optional_at(parent: &OwnedFd, name: &str) -> DispatchStoreResult<()> {
        match unlinkat(parent, name, AtFlags::empty()) {
            Ok(()) | Err(rustix::io::Errno::NOENT) => Ok(()),
            Err(source) => Err(DispatchStoreError::io(
                "remove singleton temporary",
                name.into(),
                source,
            )),
        }
    }

    fn publish_no_clobber(
        store: &DispatchStore,
        dispatch_fd: &OwnedFd,
        run: &RunId,
        target: &str,
        bytes: &[u8],
    ) -> DispatchStoreResult<()> {
        let (temp_name, mut temp) = create_temp(store, dispatch_fd, run, target)?;
        let result = (|| {
            temp.write_all(bytes).map_err(|source| {
                DispatchStoreError::io(
                    "write dispatch temp",
                    store.runs_root.join(run.as_str()).join(&temp_name),
                    source,
                )
            })?;
            temp.sync_all().map_err(|source| {
                DispatchStoreError::io(
                    "fsync dispatch temp",
                    store.runs_root.join(run.as_str()).join(&temp_name),
                    source,
                )
            })?;
            linkat(
                dispatch_fd,
                &temp_name,
                dispatch_fd,
                target,
                AtFlags::empty(),
            )
            .map_err(|source| {
                if source == rustix::io::Errno::EXIST {
                    DispatchStoreError::AlreadyExists {
                        path: store
                            .runs_root
                            .join(run.as_str())
                            .join("dispatch")
                            .join(target),
                    }
                } else {
                    DispatchStoreError::io(
                        "publish dispatch record",
                        store
                            .runs_root
                            .join(run.as_str())
                            .join("dispatch")
                            .join(target),
                        source,
                    )
                }
            })?;
            unlinkat(dispatch_fd, &temp_name, AtFlags::empty()).map_err(|source| {
                DispatchStoreError::io(
                    "unlink dispatch temp",
                    store
                        .runs_root
                        .join(run.as_str())
                        .join("dispatch")
                        .join(&temp_name),
                    source,
                )
            })?;
            fs::fsync(dispatch_fd).map_err(|source| {
                DispatchStoreError::io(
                    "fsync dispatch directory",
                    store.runs_root.join(run.as_str()).join("dispatch"),
                    source,
                )
            })
        })();
        if result.is_err() {
            let _ = unlinkat(dispatch_fd, &temp_name, AtFlags::empty());
        }
        result
    }

    fn replace_record(
        store: &DispatchStore,
        dispatch_fd: &OwnedFd,
        run: &RunId,
        record: &DispatchRecord,
    ) -> DispatchStoreResult<()> {
        replace_existing(
            store,
            dispatch_fd,
            run,
            &record_name(&record.agent_id),
            &encode_record(record)?,
        )
    }

    fn replace_existing(
        store: &DispatchStore,
        dispatch_fd: &OwnedFd,
        run: &RunId,
        target: &str,
        bytes: &[u8],
    ) -> DispatchStoreResult<()> {
        let (temp_name, mut temp) = create_temp(store, dispatch_fd, run, target)?;
        let result = (|| {
            temp.write_all(bytes).map_err(|source| {
                DispatchStoreError::io(
                    "write dispatch temp",
                    store.runs_root.join(run.as_str()).join(&temp_name),
                    source,
                )
            })?;
            temp.sync_all().map_err(|source| {
                DispatchStoreError::io(
                    "fsync dispatch temp",
                    store.runs_root.join(run.as_str()).join(&temp_name),
                    source,
                )
            })?;
            let target_path = store
                .runs_root
                .join(run.as_str())
                .join("dispatch")
                .join(target);
            let stat =
                fs::statat(dispatch_fd, target, AtFlags::SYMLINK_NOFOLLOW).map_err(|source| {
                    unsafe_path_or_io("inspect dispatch record", target_path.clone(), source)
                })?;
            if !FileType::from_raw_mode(stat.st_mode).is_file() {
                return Err(DispatchStoreError::UnsafePath { path: target_path });
            }
            renameat(dispatch_fd, &temp_name, dispatch_fd, target).map_err(|source| {
                DispatchStoreError::io(
                    "replace dispatch record",
                    store
                        .runs_root
                        .join(run.as_str())
                        .join("dispatch")
                        .join(target),
                    source,
                )
            })?;
            fs::fsync(dispatch_fd).map_err(|source| {
                DispatchStoreError::io(
                    "fsync dispatch directory",
                    store.runs_root.join(run.as_str()).join("dispatch"),
                    source,
                )
            })
        })();
        if result.is_err() {
            let _ = unlinkat(dispatch_fd, &temp_name, AtFlags::empty());
        }
        result
    }

    fn create_temp(
        store: &DispatchStore,
        dispatch_fd: &OwnedFd,
        run: &RunId,
        target: &str,
    ) -> DispatchStoreResult<(String, File)> {
        for _ in 0..MAX_TEMP_ATTEMPTS {
            let sequence = TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
            let nanos = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .map(|duration| duration.as_nanos())
                .unwrap_or(0);
            let name = format!(".{target}.{nanos:x}.{sequence:x}.tmp");
            match openat(
                dispatch_fd,
                &name,
                OFlags::WRONLY | OFlags::CREATE | OFlags::EXCL | OFlags::CLOEXEC | OFlags::NOFOLLOW,
                Mode::RUSR | Mode::WUSR,
            ) {
                Ok(fd) => return Ok((name, File::from(fd))),
                Err(rustix::io::Errno::EXIST) => continue,
                Err(source) => {
                    return Err(DispatchStoreError::io(
                        "create dispatch temp",
                        store
                            .runs_root
                            .join(run.as_str())
                            .join("dispatch")
                            .join(name),
                        source,
                    ));
                }
            }
        }
        Err(DispatchStoreError::io(
            "create dispatch temp",
            store.runs_root.join(run.as_str()).join("dispatch"),
            std::io::Error::new(
                std::io::ErrorKind::AlreadyExists,
                "temporary name collision budget exhausted",
            ),
        ))
    }

    fn encode_record(record: &DispatchRecord) -> DispatchStoreResult<Vec<u8>> {
        encode_document(record)
    }

    fn encode_document(document: &impl serde::Serialize) -> DispatchStoreResult<Vec<u8>> {
        let mut bytes = serde_json::to_vec(document).map_err(|error| {
            DispatchStoreError::Domain(DispatchError::InvalidRecord(error.to_string()))
        })?;
        bytes.push(b'\n');
        let max = usize::try_from(MAX_RECORD_BYTES).expect("record limit fits usize");
        if bytes.len() > max {
            return Err(DispatchStoreError::RecordTooLarge {
                size: bytes.len(),
                max,
            });
        }
        Ok(bytes)
    }

    fn read_bounded(file: File, limit: u64) -> std::io::Result<Vec<u8>> {
        let mut bytes = Vec::new();
        file.take(limit + 1).read_to_end(&mut bytes)?;
        if bytes.len() as u64 > limit {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "document exceeds size limit",
            ));
        }
        Ok(bytes)
    }

    fn record_name(agent_id: &AgentId) -> String {
        format!("{}.json", agent_id.as_str())
    }

    fn unknown_record(
        run: &RunId,
        agent_id: &AgentId,
        reason: impl Into<String>,
    ) -> DispatchStoreError {
        DispatchStoreError::UnknownRecord {
            run: run.clone(),
            agent_id: agent_id.clone(),
            reason: reason.into(),
        }
    }

    fn reject_parent_components(path: &Path) -> DispatchStoreResult<()> {
        if !path.is_absolute()
            || path.components().any(|component| {
                matches!(
                    component,
                    std::path::Component::ParentDir | std::path::Component::CurDir
                )
            })
        {
            return Err(DispatchStoreError::UnsafePath {
                path: path.to_path_buf(),
            });
        }
        Ok(())
    }

    fn unsafe_path_or_io(
        operation: &'static str,
        path: PathBuf,
        source: rustix::io::Errno,
    ) -> DispatchStoreError {
        if matches!(source, rustix::io::Errno::LOOP | rustix::io::Errno::NOTDIR) {
            DispatchStoreError::UnsafePath { path }
        } else {
            DispatchStoreError::io(operation, path, source)
        }
    }

    struct DispatchLock(File);

    impl Drop for DispatchLock {
        fn drop(&mut self) {
            let _ = self.0.unlock();
        }
    }
}

#[cfg(not(unix))]
mod platform {
    //! The non-unix twin of the descriptor-anchored dispatch ledger.
    //!
    //! Every ledger operation performs the SAME sequence as the unix module --
    //! validate the runs root, resolve the run directory, resolve the dispatch
    //! directory, take the exclusive dispatch lock, then operate -- and returns
    //! the SAME error variants, because callers and the hook fixtures branch on
    //! them. `AlreadyExists` in particular is the fact that stops a second
    //! `SessionStart` from silently overwriting a live root binding.
    //!
    //! The anchoring differs, and only the anchoring: paths with per-component
    //! link rejection instead of a chain of directory descriptors. See
    //! `crate::safe_fs` for exactly what that does and does not guarantee.
    //!
    //! Active-run resolution is deliberately NOT twinned. Enumerating the runs
    //! root, filtering names, sorting, and choosing among the candidates is
    //! platform-free, so it lives once in `super::resolve_active_run` and this
    //! module contributes only the anchoring residue that resolver calls:
    //! `open_runs_root`, `run_names`, and `read_run_document`. A twin of the choosing logic
    //! is a twin that drifts, because only one of the two is ever compiled
    //! here.

    use std::fs::{File, TryLockError};
    use std::io::Write;

    use super::*;
    use crate::safe_fs;

    const MAX_RECORD_BYTES: u64 = 1_048_576;
    const LOCK_RETRY_INTERVAL: Duration = Duration::from_millis(10);

    pub(super) fn publish_singleton(
        store: &DispatchStore,
        publication: &DispatchSingletonPublication,
    ) -> DispatchStoreResult<()> {
        let run = RunId::new(publication.run_id.clone()).map_err(DispatchStoreError::Domain)?;
        let dispatch = dispatch_dir(store, &run, true)?;
        let _lock = acquire_lock(store, &run)?;
        let target = dispatch.join(publication_target_name(publication)?);
        let temp = dispatch.join(preparing_name(&publication.nonce));
        let bytes = publication.record_json.as_bytes();
        match safe_fs::write_no_clobber(&temp, bytes) {
            Ok(true) => {}
            Ok(false) => {
                let existing =
                    safe_fs::read_regular_nofollow(&temp, MAX_RECORD_BYTES).map_err(|source| {
                        DispatchStoreError::io(
                            "read singleton preparing file",
                            temp.clone(),
                            source,
                        )
                    })?;
                if existing != bytes {
                    return Err(DispatchStoreError::SingletonPublication {
                        nonce: publication.nonce.clone(),
                        reason: "preparing file already contains different bytes".into(),
                    });
                }
            }
            Err(source) => {
                return Err(DispatchStoreError::io(
                    "prepare singleton publication",
                    temp,
                    source,
                ));
            }
        }
        if store.publication_fault == Some(PublicationFault::AfterPreparing) {
            return Err(DispatchStoreError::SingletonPublication {
                nonce: publication.nonce.clone(),
                reason: "injected failure after preparing filesystem bytes".into(),
            });
        }
        publish_final_no_clobber(store, &run, &temp, &target, bytes)
    }

    pub(super) fn publish_review_replacement_prepared(
        store: &DispatchStore,
        publication: &DispatchSingletonPublication,
        registry: &mut Registry,
    ) -> DispatchStoreResult<()> {
        let run = RunId::new(&publication.run_id)?;
        let dispatch = dispatch_dir(store, &run, false)?;
        let _lock = acquire_lock(store, &run)?;
        if read_optional_path(&dispatch.join(publication_target_name(publication)?))?.as_deref()
            != Some(publication.record_json.as_bytes())
        {
            return Err(DispatchStoreError::Reconciliation(
                "replacement filesystem bytes differ from the prepared publication".into(),
            ));
        }
        publish_review_replacement_at(
            &LockedDispatchRun {
                store,
                run: &run,
                dispatch: &dispatch,
            },
            publication,
            registry,
        )
    }

    pub(super) fn reconcile_singleton(
        store: &DispatchStore,
        publication: &DispatchSingletonPublication,
        registry: &mut Registry,
    ) -> DispatchStoreResult<PublicationReconcile> {
        let run = RunId::new(publication.run_id.clone()).map_err(DispatchStoreError::Domain)?;
        let dispatch = dispatch_dir(store, &run, false)?;
        let _lock = acquire_lock(store, &run)?;
        let target = dispatch.join(publication_target_name(publication)?);
        let temp = dispatch.join(preparing_name(&publication.nonce));
        let expected = publication.record_json.as_bytes();
        let target_bytes = read_optional_path(&target)?;
        match publication.state {
            SingletonPublicationState::Quarantined => Ok(PublicationReconcile::Unchanged),
            SingletonPublicationState::Published => {
                if target_bytes.as_deref() == Some(expected) {
                    Ok(PublicationReconcile::Unchanged)
                } else if reconcile_review_terminal_publication(
                    &LockedDispatchRun {
                        store,
                        run: &run,
                        dispatch: &dispatch,
                    },
                    publication,
                    target_bytes.as_deref(),
                    registry,
                )? {
                    Ok(PublicationReconcile::Refreshed)
                } else {
                    Ok(PublicationReconcile::Quarantined(
                        "published filesystem bytes are missing or corrupt".into(),
                    ))
                }
            }
            SingletonPublicationState::Preparing => {
                let temp_bytes = read_optional_path(&temp)?;
                if target_bytes
                    .as_deref()
                    .is_some_and(|value| value != expected)
                    || temp_bytes.as_deref().is_some_and(|value| value != expected)
                {
                    return Ok(PublicationReconcile::Quarantined(
                        "preparing or published filesystem bytes are corrupt".into(),
                    ));
                }
                if target_bytes.as_deref() == Some(expected) {
                    finish_preparing_review_replacement(
                        &LockedDispatchRun {
                            store,
                            run: &run,
                            dispatch: &dispatch,
                        },
                        publication,
                        registry,
                    )?;
                    let _ = safe_fs::remove_file_nofollow(&temp);
                    return Ok(PublicationReconcile::Published);
                }
                if temp_bytes.as_deref() == Some(expected) {
                    publish_final_no_clobber(store, &run, &temp, &target, expected)?;
                    finish_preparing_review_replacement(
                        &LockedDispatchRun {
                            store,
                            run: &run,
                            dispatch: &dispatch,
                        },
                        publication,
                        registry,
                    )?;
                    return Ok(PublicationReconcile::Published);
                }
                Ok(PublicationReconcile::Quarantined(
                    "preparing publication has no durable filesystem bytes".into(),
                ))
            }
        }
    }

    pub(super) fn quarantine_singleton(
        store: &DispatchStore,
        publication: &DispatchSingletonPublication,
    ) -> DispatchStoreResult<()> {
        let run = RunId::new(publication.run_id.clone()).map_err(DispatchStoreError::Domain)?;
        let dispatch = store.runs_root.join(run.as_str()).join("dispatch");
        let _lock = acquire_lock(store, &run)?;
        crate::safe_fs::reject_link_components(&dispatch).map_err(|_| {
            DispatchStoreError::UnsafePath {
                path: dispatch.clone(),
            }
        })?;
        let quarantine = dispatch.join("quarantine");
        safe_fs::ensure_directory(&dispatch, "quarantine").map_err(|source| {
            DispatchStoreError::io(
                "create singleton quarantine directory",
                quarantine.clone(),
                source,
            )
        })?;
        let temp = dispatch.join(preparing_name(&publication.nonce));
        let preparing_target = quarantine.join(format!("{}.preparing", publication.nonce));
        if read_optional_path(&temp)?.is_some() {
            if !read_optional_path(&preparing_target)?.is_some() {
                std::fs::rename(&temp, &preparing_target).map_err(|source| {
                    DispatchStoreError::io(
                        "quarantine singleton preparing file",
                        preparing_target.clone(),
                        source,
                    )
                })?;
            } else {
                safe_fs::remove_file_nofollow(&temp).map_err(|source| {
                    DispatchStoreError::io(
                        "remove duplicate singleton preparing file",
                        temp.clone(),
                        source,
                    )
                })?;
            }
        } else if read_optional_path(&preparing_target)?.is_none() {
            safe_fs::write_no_clobber(&preparing_target, publication.record_json.as_bytes())
                .map_err(|source| {
                    DispatchStoreError::io(
                        "write singleton quarantine marker",
                        preparing_target.clone(),
                        source,
                    )
                })?;
        }
        let target = dispatch.join(publication_target_name(publication)?);
        let published_target = quarantine.join(format!("{}.published", publication.nonce));
        if read_optional_path(&target)?.is_some() {
            if read_optional_path(&published_target)?.is_none() {
                std::fs::rename(&target, &published_target).map_err(|source| {
                    DispatchStoreError::io(
                        "quarantine singleton published file",
                        published_target.clone(),
                        source,
                    )
                })?;
            } else {
                safe_fs::remove_file_nofollow(&target).map_err(|source| {
                    DispatchStoreError::io(
                        "remove duplicate singleton published file",
                        target.clone(),
                        source,
                    )
                })?;
            }
        }
        Ok(())
    }

    /// The anchor the shared resolver holds across enumeration.
    ///
    /// The unix twin hands back a directory descriptor that every later
    /// `openat` resolves against. There is no descriptor here -- anchoring is
    /// per-component and re-checked at each open -- so this carries no data
    /// and exists only as proof that the runs root was validated before a
    /// single entry was enumerated. Only [`open_runs_root`] can mint one.
    pub(super) struct RunsRoot;

    /// Validate the runs root and mint the resolver's anchor.
    pub(super) fn open_runs_root(store: &DispatchStore) -> DispatchStoreResult<RunsRoot> {
        reject_parent_components(&store.runs_root)?;
        Ok(RunsRoot)
    }

    pub(super) fn run_names(
        store: &DispatchStore,
        _root: &RunsRoot,
    ) -> DispatchStoreResult<Vec<String>> {
        let entries = std::fs::read_dir(&store.runs_root).map_err(|source| {
            DispatchStoreError::io("read runs directory", store.runs_root.clone(), source)
        })?;
        let mut names = Vec::new();
        for entry in entries {
            let entry = entry.map_err(|source| {
                DispatchStoreError::io("read runs directory", store.runs_root.clone(), source)
            })?;
            let name = entry.file_name();
            let Some(name) = name.to_str() else {
                return Err(DispatchStoreError::UnsafePath {
                    path: store.runs_root.join("<non-utf8>"),
                });
            };
            names.push(name.to_owned());
        }
        Ok(names)
    }

    pub(super) fn publish_root_binding(
        store: &DispatchStore,
        binding: &RootSessionBinding,
    ) -> DispatchStoreResult<()> {
        let dispatch = dispatch_dir(store, &binding.run, true)?;
        let _lock = acquire_lock(store, &binding.run)?;
        let bytes = encode_document(binding)?;
        publish_no_clobber(
            &dispatch.join(root_binding_name(&binding.session_id)),
            &bytes,
        )
    }

    pub(super) fn load_current_root_binding(
        store: &DispatchStore,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        ensure_no_root_renewal(store, session_id)?;
        read_current_root_binding_raw(store, session_id)
    }

    fn read_current_root_binding_raw(
        store: &DispatchStore,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        let path = store.current_root_binding_path(session_id);
        let bytes = read_document(&path, "read current root binding")?;
        decode_current_root_binding(&bytes, session_id)
    }

    pub(super) fn root_renewal_intent(
        store: &DispatchStore,
        session: &SessionId,
    ) -> DispatchStoreResult<Option<RootRenewalIntent>> {
        open_runs_root(store)?;
        read_optional_path(&store.runs_root.join(root_renewal_name(session)))?
            .map(|bytes| decode_root_renewal(&bytes, session))
            .transpose()
    }

    fn ensure_no_root_renewal(
        store: &DispatchStore,
        session: &SessionId,
    ) -> DispatchStoreResult<()> {
        if root_renewal_intent(store, session)?.is_some() {
            return Err(root_renewal_error(
                "root lease renewal is incomplete; repeat the exact explicit bind-root request",
            ));
        }
        Ok(())
    }

    pub(super) fn commit_root_renewal(
        store: &DispatchStore,
        intent: &RootRenewalIntent,
        recover: bool,
    ) -> DispatchStoreResult<RootSessionBinding> {
        open_runs_root(store)?;
        let _current_lock = acquire_current_root_lock(store)?;
        let run = &intent.expected.run;
        let session = &intent.expected.session_id;
        dispatch_dir(store, run, false)?;
        let _run_lock = acquire_lock(store, run)?;
        let intent_path = store.runs_root.join(root_renewal_name(session));
        let existing_intent = root_renewal_intent(store, session)?;
        if (recover && existing_intent.as_ref() != Some(intent))
            || (!recover && existing_intent.is_some())
        {
            return Err(root_renewal_error(
                "root renewal intent changed or requires explicit same-root recovery",
            ));
        }
        let binding_path = store.root_binding_path(run, session);
        let binding = decode_root_binding(
            &read_document(&binding_path, "read root binding")?,
            run,
            session,
        )?;
        let index = read_current_root_binding_raw(store, session)?;
        let profile = read_optional_path(&store.profile_lease_path(run, session))?
            .map(|bytes| decode_profile_lease(&bytes, run, session))
            .transpose()?;
        let state = load_run(store, run)?;
        let (run_old, index_old) = validate_root_renewal_observation(
            intent,
            &binding,
            &index,
            profile.as_ref(),
            &state.status,
        )?;
        if !recover && !(run_old && index_old) {
            return Err(root_renewal_error(
                "root renewal expected authority is no longer current",
            ));
        }
        if !recover {
            publish_root_renewal_intent(store, &intent_path, &encode_document(intent)?)?;
            store.root_renewal_fault(RootRenewalFault::AfterIntentCommit)?;
        }
        let bytes = encode_document(&intent.replacement)?;
        if run_old {
            safe_fs::replace_atomic(&binding_path, &bytes).map_err(|source| {
                DispatchStoreError::io("renew root binding", binding_path.clone(), source)
            })?;
            store.root_renewal_fault(RootRenewalFault::AfterRunBindingCommit)?;
        }
        let index_path = store.current_root_binding_path(session);
        if index_old {
            safe_fs::replace_atomic(&index_path, &bytes).map_err(|source| {
                DispatchStoreError::io("renew current root binding", index_path.clone(), source)
            })?;
            store.root_renewal_fault(RootRenewalFault::AfterIndexCommit)?;
        }
        if decode_root_binding(
            &read_document(&binding_path, "read renewed root binding")?,
            run,
            session,
        )? != intent.replacement
            || read_current_root_binding_raw(store, session)? != intent.replacement
        {
            return Err(root_renewal_error(
                "root renewal records changed before commit",
            ));
        }
        safe_fs::remove_file_nofollow(&intent_path)
            .map_err(|source| DispatchStoreError::io("commit root renewal", intent_path, source))?;
        Ok(intent.replacement.clone())
    }

    fn publish_root_renewal_intent(
        store: &DispatchStore,
        path: &Path,
        bytes: &[u8],
    ) -> DispatchStoreResult<()> {
        safe_fs::reject_link_components(path).map_err(|source| {
            DispatchStoreError::io("anchor root renewal intent", path.to_path_buf(), source)
        })?;
        let temporary = store
            .runs_root
            .join(format!(".root-renewal.{}.tmp", uuid::Uuid::now_v7()));
        let mut file = std::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&temporary)
            .map_err(|source| {
                DispatchStoreError::io("create root renewal intent", temporary.clone(), source)
            })?;
        file.write_all(bytes).map_err(|source| {
            DispatchStoreError::io("write root renewal intent", temporary.clone(), source)
        })?;
        file.sync_all().map_err(|source| {
            DispatchStoreError::io("fsync root renewal intent", temporary.clone(), source)
        })?;
        drop(file);
        if let Err(source) = rename_root_renewal_intent(&temporary, path) {
            let _ = safe_fs::remove_file_nofollow(&temporary);
            return Err(DispatchStoreError::io(
                "publish root renewal intent",
                path.to_path_buf(),
                source,
            ));
        }
        store.root_renewal_fault(RootRenewalFault::AfterIntentPublish)
    }

    #[cfg(windows)]
    #[allow(unsafe_code)] // Reviewed native OS boundary; no replacement or cross-volume copy.
    fn rename_root_renewal_intent(temporary: &Path, path: &Path) -> std::io::Result<()> {
        use std::os::windows::ffi::OsStrExt;
        use windows_sys::Win32::Storage::FileSystem::{MOVEFILE_WRITE_THROUGH, MoveFileExW};
        let temporary: Vec<u16> = temporary.as_os_str().encode_wide().chain(Some(0)).collect();
        let path: Vec<u16> = path.as_os_str().encode_wide().chain(Some(0)).collect();
        // SAFETY: both buffers are live, NUL-terminated filesystem paths. No
        // REPLACE_EXISTING, COPY_ALLOWED or delayed operation is authorized.
        if unsafe { MoveFileExW(temporary.as_ptr(), path.as_ptr(), MOVEFILE_WRITE_THROUGH) } == 0 {
            Err(std::io::Error::last_os_error())
        } else {
            Ok(())
        }
    }

    #[cfg(not(windows))]
    fn rename_root_renewal_intent(_temporary: &Path, _path: &Path) -> std::io::Result<()> {
        Err(std::io::Error::new(
            std::io::ErrorKind::Unsupported,
            "atomic no-replace root renewal publication is unavailable on this platform",
        ))
    }

    pub(super) fn activate_current_root_binding(
        store: &DispatchStore,
        binding: &RootSessionBinding,
    ) -> DispatchStoreResult<()> {
        let _lock = acquire_current_root_lock(store)?;
        let path = store.current_root_binding_path(&binding.session_id);
        let bytes = encode_document(binding)?;
        match load_current_root_binding(store, &binding.session_id) {
            Ok(current) if current == *binding => Ok(()),
            Ok(current) => {
                validate_current_binding_transition(&current, binding)?;
                safe_fs::replace_atomic(&path, &bytes).map_err(|source| {
                    DispatchStoreError::io("replace current root binding", path.clone(), source)
                })
            }
            Err(error) if is_not_found(&error) => match safe_fs::write_no_clobber(&path, &bytes) {
                Ok(true) => Ok(()),
                Ok(false) => Err(DispatchStoreError::AlreadyExists { path }),
                Err(source) => Err(DispatchStoreError::io(
                    "publish current root binding",
                    path,
                    source,
                )),
            },
            Err(error) => Err(error),
        }
    }

    pub(super) fn transition_root_binding_to_execution(
        store: &DispatchStore,
        expected: &RootSessionBinding,
        replacement: &RootSessionBinding,
    ) -> DispatchStoreResult<()> {
        let dispatch = dispatch_dir(store, &replacement.run, false)?;
        let _lock = acquire_lock(store, &replacement.run)?;
        ensure_no_root_renewal(store, &replacement.session_id)?;
        let state = load_run(store, &replacement.run)?;
        if !state.status.is(RunStatus::Executing) {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidRecord(
                "root binding cannot enter execution before the selected run is executing".into(),
            )));
        }
        let path = store.root_binding_path(&replacement.run, &replacement.session_id);
        let current = decode_root_binding(
            &read_document(&path, "read root binding")?,
            &replacement.run,
            &replacement.session_id,
        )?;
        if &current != expected {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidRecord(
                "root binding changed during mode transition".into(),
            )));
        }
        safe_fs::replace_atomic(&path, &encode_document(replacement)?).map_err(|source| {
            DispatchStoreError::io("replace root binding", path.clone(), source)
        })?;
        let _ = dispatch;
        Ok(())
    }

    pub(super) fn publish_profile_lease(
        store: &DispatchStore,
        lease: &ProfileLease,
    ) -> DispatchStoreResult<()> {
        let dispatch = dispatch_dir(store, &lease.run, true)?;
        let _lock = acquire_lock(store, &lease.run)?;
        publish_no_clobber(
            &dispatch.join(profile_lease_name(&lease.root_session_id)),
            &encode_document(lease)?,
        )
    }

    pub(super) fn replace_profile_lease(
        store: &DispatchStore,
        expected: &ProfileLease,
        replacement: &ProfileLease,
    ) -> DispatchStoreResult<()> {
        let dispatch = dispatch_dir(store, &replacement.run, false)?;
        let _lock = acquire_lock(store, &replacement.run)?;
        let path = store.profile_lease_path(&replacement.run, &replacement.root_session_id);
        let current = decode_profile_lease(
            &read_document(&path, "read profile lease")?,
            &replacement.run,
            &replacement.root_session_id,
        )?;
        if &current != expected {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidProfile(
                "profile lease changed during state transition".into(),
            )));
        }
        safe_fs::replace_atomic(&path, &encode_document(replacement)?)
            .map_err(|source| DispatchStoreError::io("replace profile lease", path, source))?;
        let _ = dispatch;
        Ok(())
    }

    pub(super) fn publish_review_custody(
        store: &DispatchStore,
        custody: &ReviewCustody,
    ) -> DispatchStoreResult<()> {
        let dispatch = dispatch_dir(store, &custody.run, true)?;
        let _lock = acquire_lock(store, &custody.run)?;
        publish_no_clobber(
            &dispatch.join(review_custody_name(&custody.subject_agent_id)),
            &encode_document(custody)?,
        )
    }

    pub(super) fn replace_review_custody(
        store: &DispatchStore,
        expected: &ReviewCustody,
        replacement: &ReviewCustody,
    ) -> DispatchStoreResult<()> {
        let dispatch = dispatch_dir(store, &replacement.run, false)?;
        let _lock = acquire_lock(store, &replacement.run)?;
        let path = store.review_custody_path(&replacement.run, &replacement.subject_agent_id);
        let current = decode_review_custody(
            &read_document(&path, "read review custody")?,
            &replacement.run,
            &replacement.subject_agent_id,
        )?;
        if &current != expected {
            return Err(DispatchStoreError::Domain(
                DispatchError::InvalidReviewCustody(
                    "review custody changed during state transition".into(),
                ),
            ));
        }
        safe_fs::replace_atomic(&path, &encode_document(replacement)?)
            .map_err(|source| DispatchStoreError::io("replace review custody", path, source))?;
        let _ = dispatch;
        Ok(())
    }

    pub(super) fn publish_skill_use(
        store: &DispatchStore,
        challenge: &SkillUseChallenge,
    ) -> DispatchStoreResult<()> {
        let dispatch = dispatch_dir(store, &challenge.run, true)?;
        let _lock = acquire_lock(store, &challenge.run)?;
        let key = SkillUseKey::from_challenge(challenge)?;
        validate_skill_use_root(store, &challenge.run, key)?;
        publish_no_clobber(
            &dispatch.join(skill_use_name(key, &challenge.skill)?),
            &encode_document(challenge)?,
        )
    }

    pub(super) fn replace_skill_use(
        store: &DispatchStore,
        expected: &SkillUseChallenge,
        replacement: &SkillUseChallenge,
    ) -> DispatchStoreResult<()> {
        let dispatch = dispatch_dir(store, &replacement.run, false)?;
        let _lock = acquire_lock(store, &replacement.run)?;
        let path = store.skill_use_path(replacement)?;
        let key = SkillUseKey::from_challenge(replacement)?;
        validate_skill_use_root(store, &replacement.run, key)?;
        let current = decode_skill_use(
            &read_document(&path, "read skill-use challenge")?,
            &replacement.run,
            key,
            &replacement.skill,
        )?;
        if &current != expected {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidSkillUse(
                "skill-use challenge changed during state transition".into(),
            )));
        }
        safe_fs::replace_atomic(&path, &encode_document(replacement)?).map_err(|source| {
            DispatchStoreError::io("replace skill-use challenge", path, source)
        })?;
        let _ = dispatch;
        Ok(())
    }

    pub(super) fn publish_pending<F>(
        store: &DispatchStore,
        pending: &PendingDispatch,
        lease_ms: Option<u64>,
        authorize: F,
    ) -> DispatchStoreResult<PendingDispatch>
    where
        F: FnOnce(&PendingDispatch, i64, &LockedDispatchRun<'_>) -> DispatchStoreResult<()>,
    {
        let dispatch = dispatch_dir(store, &pending.run, true)?;
        let _lock = acquire_lock(store, &pending.run)?;
        let state = load_run(store, &pending.run)?;
        if matches!(
            state.status.known(),
            Some(RunStatus::Closed | RunStatus::Closing)
        ) {
            return Err(DispatchStoreError::PendingPath {
                run: pending.run.clone(),
            });
        }
        if state.status != pending.run_status {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidPending(
                "run state changed during pending publication".into(),
            )));
        }
        let agent_id = &pending.expected_attachment.agent_id;
        match read_record(store, &pending.run, agent_id) {
            Ok(_) => {
                return Err(DispatchStoreError::AlreadyExists {
                    path: store.record_path(&pending.run, agent_id),
                });
            }
            Err(DispatchStoreError::UnknownRecord { reason, .. })
                if reason == "record is absent" => {}
            Err(error) => return Err(error),
        }
        let pending = refresh_pending_lease(pending, lease_ms)?;
        let authority = LockedDispatchRun {
            store,
            run: &pending.run,
            dispatch: &dispatch,
        };
        authorize(&pending, current_time_millis(), &authority)?;
        let bytes = encode_document(&pending)?;
        publish_no_clobber(&dispatch.join(pending_name(pending.launch_id_hash)), &bytes)?;
        Ok(pending)
    }

    pub(super) fn load_inventory(
        store: &DispatchStore,
        run: &RunId,
    ) -> DispatchStoreResult<DispatchInventory> {
        let dispatch = dispatch_dir(store, run, false)?;
        let _lock = acquire_lock(store, run)?;
        read_locked_inventory(&LockedDispatchRun {
            store,
            run,
            dispatch: &dispatch,
        })
    }

    pub(super) fn with_run_access<T>(
        store: &DispatchStore,
        run: &RunId,
        access: &crate::run_store::RunAccess<'_>,
        operation: impl FnOnce(&LockedDispatchRun<'_>) -> DispatchStoreResult<T>,
    ) -> DispatchStoreResult<T> {
        if access.run_path != store.runs_root.join(run.as_str()) {
            return Err(DispatchStoreError::UnsafePath {
                path: store.runs_root.join(run.as_str()),
            });
        }
        let dispatch = dispatch_dir(store, run, false)?;
        operation(&LockedDispatchRun {
            store,
            run,
            dispatch: &dispatch,
        })
    }

    pub(super) fn read_locked_inventory(
        authority: &LockedDispatchRun<'_>,
    ) -> DispatchStoreResult<DispatchInventory> {
        use std::os::windows::fs::OpenOptionsExt;
        let open_directory = || {
            safe_fs::reject_link_components(authority.dispatch)?;
            std::fs::OpenOptions::new()
                .read(true)
                .custom_flags(0x0020_0000 | 0x0200_0000)
                .open(authority.dispatch)
        };
        let anchor = open_directory().map_err(|source| {
            DispatchStoreError::io(
                "anchor dispatch inventory",
                authority.dispatch.to_path_buf(),
                source,
            )
        })?;
        let identity = safe_fs::windows_file_id(&anchor).map_err(|source| {
            DispatchStoreError::io(
                "identify dispatch inventory",
                authority.dispatch.to_path_buf(),
                source,
            )
        })?;
        let mut inventory = DispatchInventory::default();
        for entry in std::fs::read_dir(authority.dispatch).map_err(|source| {
            DispatchStoreError::io(
                "read dispatch inventory",
                authority.dispatch.to_path_buf(),
                source,
            )
        })? {
            let entry = entry.map_err(|source| {
                DispatchStoreError::io(
                    "read dispatch inventory",
                    authority.dispatch.to_path_buf(),
                    source,
                )
            })?;
            let name =
                entry
                    .file_name()
                    .into_string()
                    .map_err(|_| DispatchStoreError::UnsafePath {
                        path: authority.dispatch.join("<non-utf8>"),
                    })?;
            match inventory_entry(&name)? {
                Some(InventoryEntry::Record(agent)) => {
                    inventory
                        .records
                        .push(read_record(authority.store, authority.run, &agent)?)
                }
                Some(InventoryEntry::Pending(hash)) => {
                    inventory
                        .pending
                        .push(read_pending(authority.store, authority.run, hash)?)
                }
                None => {}
            }
        }
        let current = open_directory()
            .and_then(|file| safe_fs::windows_file_id(&file))
            .map_err(|source| {
                DispatchStoreError::io(
                    "recheck dispatch inventory",
                    authority.dispatch.to_path_buf(),
                    source,
                )
            })?;
        if current != identity {
            return Err(DispatchStoreError::UnsafePath {
                path: authority.dispatch.to_path_buf(),
            });
        }
        sort_inventory(&mut inventory);
        Ok(inventory)
    }

    pub(super) fn load_pending(
        store: &DispatchStore,
        run: &RunId,
        launch_id_hash: [u8; 32],
    ) -> DispatchStoreResult<PendingDispatch> {
        // Same mapping as the unix twin: an absent dispatch directory means no
        // such pending launch, not an I/O fault. Without this the non-unix side
        // reported a raw io error and a missing launch looked like a failure to
        // read rather than a launch that was never claimed.
        dispatch_dir(store, run, false).map_err(|error| {
            if is_not_found(&error) {
                DispatchStoreError::PendingNotFound { run: run.clone() }
            } else {
                error
            }
        })?;
        let _lock = acquire_lock(store, run)?;
        read_pending(store, run, launch_id_hash)
    }

    fn read_pending(
        store: &DispatchStore,
        run: &RunId,
        launch_id_hash: [u8; 32],
    ) -> DispatchStoreResult<PendingDispatch> {
        let path = store
            .runs_root
            .join(run.as_str())
            .join("dispatch")
            .join(pending_name(launch_id_hash));
        let bytes = read_document(&path, "read pending dispatch").map_err(|error| {
            if is_not_found(&error) {
                DispatchStoreError::PendingNotFound { run: run.clone() }
            } else {
                error
            }
        })?;
        let pending: PendingDispatch = serde_json::from_slice(&bytes).map_err(|error| {
            DispatchStoreError::Domain(DispatchError::InvalidRecord(error.to_string()))
        })?;
        pending.validate()?;
        if pending.run != *run || !constant_time_digest_eq(&pending.launch_id_hash, &launch_id_hash)
        {
            return Err(DispatchStoreError::PendingPath { run: run.clone() });
        }
        Ok(pending)
    }

    pub(super) fn load_pending_for_agent(
        store: &DispatchStore,
        run: &RunId,
        agent_id: &AgentId,
    ) -> DispatchStoreResult<PendingDispatch> {
        let dispatch = dispatch_dir(store, run, false)?;
        let _lock = acquire_lock(store, run)?;
        let mut matched = None;
        for entry in std::fs::read_dir(&dispatch).map_err(|source| {
            DispatchStoreError::io("read pending launches", dispatch.clone(), source)
        })? {
            let entry = entry.map_err(|source| {
                DispatchStoreError::io("read pending launches", dispatch.clone(), source)
            })?;
            let name = entry.file_name().to_string_lossy().into_owned();
            let Some(hash) = name
                .strip_prefix("pending-")
                .and_then(|value| value.strip_suffix(".json"))
                .and_then(|value| parse_hash_name(value).ok())
            else {
                continue;
            };
            let path = dispatch.join(&name);
            let pending: PendingDispatch =
                serde_json::from_slice(&read_document(&path, "read pending dispatch")?).map_err(
                    |error| {
                        DispatchStoreError::Domain(DispatchError::InvalidRecord(error.to_string()))
                    },
                )?;
            pending.validate()?;
            if pending.run != *run || pending.launch_id_hash != hash {
                return Err(DispatchStoreError::PendingPath { run: run.clone() });
            }
            if pending.expected_attachment.agent_id == *agent_id
                && matched.replace(pending).is_some()
            {
                return Err(DispatchStoreError::Domain(DispatchError::InvalidPending(
                    "multiple pending claims name one subject agent".into(),
                )));
            }
        }
        matched.ok_or_else(|| unknown_record(run, agent_id, "pending claim is absent"))
    }

    pub(super) fn claim_pending_unspawned<F>(
        store: &DispatchStore,
        run: &RunId,
        launch_id_hash: [u8; 32],
        child_process_hash: [u8; 32],
        validate: F,
    ) -> DispatchStoreResult<PendingDispatch>
    where
        F: FnOnce(&PendingDispatch, i64, &LockedDispatchRun<'_>) -> DispatchStoreResult<()>,
    {
        let dispatch = dispatch_dir(store, run, false)?;
        let _lock = acquire_lock(store, run)?;
        let now = current_time_millis();
        let state = load_run(store, run)?;
        if matches!(
            state.status.known(),
            Some(RunStatus::Closed | RunStatus::Closing)
        ) {
            return Err(DispatchStoreError::PendingPath { run: run.clone() });
        }
        let path = dispatch.join(pending_name(launch_id_hash));
        let bytes = read_document(&path, "read pending dispatch")?;
        let pending: PendingDispatch = serde_json::from_slice(&bytes).map_err(|error| {
            DispatchStoreError::Domain(DispatchError::InvalidRecord(error.to_string()))
        })?;
        pending.validate()?;
        if state.status != pending.run_status {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidPending(
                "run state changed during pending claim".into(),
            )));
        }
        let authority = LockedDispatchRun {
            store,
            run,
            dispatch: &dispatch,
        };
        validate(&pending, now, &authority)?;
        pending.check_lease(now)?;
        let mut claimed = pending;
        claimed.claim(now, child_process_hash)?;
        safe_fs::replace_atomic(&path, &encode_document(&claimed)?).map_err(|source| {
            DispatchStoreError::io("claim pending dispatch", path.clone(), source)
        })?;
        Ok(claimed)
    }

    pub(super) fn activate_pending<F>(
        store: &DispatchStore,
        run: &RunId,
        launch_id_hash: [u8; 32],
        child_process_hash: [u8; 32],
        activate: F,
    ) -> DispatchStoreResult<DispatchRecord>
    where
        F: FnOnce(
            &PendingDispatch,
            i64,
            &LockedDispatchRun<'_>,
        ) -> DispatchStoreResult<DispatchRecord>,
    {
        let dispatch = dispatch_dir(store, run, false)?;
        let _lock = acquire_lock(store, run)?;
        let now = current_time_millis();
        let state = load_run(store, run)?;
        if matches!(
            state.status.known(),
            Some(RunStatus::Closed | RunStatus::Closing)
        ) {
            return Err(DispatchStoreError::PendingPath { run: run.clone() });
        }
        let path = dispatch.join(pending_name(launch_id_hash));
        let bytes = read_document(&path, "read pending dispatch")?;
        let pending: PendingDispatch = serde_json::from_slice(&bytes).map_err(|error| {
            DispatchStoreError::Domain(DispatchError::InvalidRecord(error.to_string()))
        })?;
        pending.validate()?;
        if state.status != pending.run_status {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidPending(
                "run state changed during pending activation".into(),
            )));
        }
        if pending.launch_state != shepherd::dispatch::PendingLaunchState::ClaimedUnspawned
            || pending.child_process_hash.as_ref() != Some(&child_process_hash)
        {
            return Err(DispatchStoreError::Domain(
                DispatchError::PendingLaunchConsumed,
            ));
        }
        let authority = LockedDispatchRun {
            store,
            run,
            dispatch: &dispatch,
        };
        let record = activate(&pending, now, &authority)?;
        let mut active = pending;
        active.activate(now, child_process_hash)?;
        publish_no_clobber(
            &dispatch.join(record_name(&record.agent_id)),
            &encode_record(&record)?,
        )?;
        safe_fs::replace_atomic(&path, &encode_document(&active)?).map_err(|source| {
            DispatchStoreError::io("activate pending dispatch", path.clone(), source)
        })?;
        Ok(record)
    }

    pub(super) fn reconcile_unspawned(
        store: &DispatchStore,
        run: &RunId,
        requested_now: Option<i64>,
    ) -> DispatchStoreResult<usize> {
        let dispatch = dispatch_dir(store, run, false)?;
        let _lock = acquire_lock(store, run)?;
        let now = requested_now.unwrap_or_else(current_time_millis);
        let path = dispatch.clone();
        let mut expired = 0;
        for entry in std::fs::read_dir(&path).map_err(|source| {
            DispatchStoreError::io("read pending launches", path.clone(), source)
        })? {
            let entry = entry.map_err(|source| {
                DispatchStoreError::io("read pending launches", path.clone(), source)
            })?;
            let name = entry.file_name().to_string_lossy().into_owned();
            let Some(hex_hash) = name
                .strip_prefix("pending-")
                .and_then(|value| value.strip_suffix(".json"))
            else {
                continue;
            };
            let Ok(launch_hash) = parse_hash_name(hex_hash) else {
                continue;
            };
            let pending_path = path.join(&name);
            let bytes = read_document(&pending_path, "read pending dispatch")?;
            let mut pending: PendingDispatch = serde_json::from_slice(&bytes).map_err(|error| {
                DispatchStoreError::Domain(DispatchError::InvalidRecord(error.to_string()))
            })?;
            pending.validate()?;
            if !constant_time_digest_eq(&pending.launch_id_hash, &launch_hash) {
                return Err(DispatchStoreError::PendingPath { run: run.clone() });
            }
            let agent_id = pending.expected_attachment.agent_id.clone();
            let active_record = match read_record(store, run, &agent_id) {
                Ok(record) if record.state == DispatchState::Active => Some(record),
                Ok(_) => None,
                Err(DispatchStoreError::UnknownRecord { reason, .. })
                    if reason == "record is absent" =>
                {
                    None
                }
                Err(error) => return Err(error),
            };
            let should_reconcile = match pending.launch_state {
                PendingLaunchState::ClaimedUnspawned => true,
                PendingLaunchState::Pending => now >= pending.expires_at,
                PendingLaunchState::Active => active_record.is_none(),
                PendingLaunchState::Quarantined
                | PendingLaunchState::LaunchFailed
                | PendingLaunchState::Canceled
                | PendingLaunchState::Expired => false,
            };
            if should_reconcile {
                if pending.launch_state == PendingLaunchState::ClaimedUnspawned
                    || (pending.launch_state == PendingLaunchState::Pending
                        && now >= pending.expires_at)
                {
                    if active_record.is_some() {
                        remove_record(store, run, &agent_id)?;
                    }
                    pending.expire()?;
                } else {
                    pending.fail_after_recovery()?;
                }
                safe_fs::replace_atomic(&pending_path, &encode_document(&pending)?).map_err(
                    |source| {
                        DispatchStoreError::io(
                            "reconcile pending dispatch",
                            pending_path.clone(),
                            source,
                        )
                    },
                )?;
                expired += 1;
            }
        }
        Ok(expired)
    }

    pub(super) fn cleanup_pending(
        store: &DispatchStore,
        run: &RunId,
        launch_hash: [u8; 32],
        state: PendingLaunchState,
    ) -> DispatchStoreResult<LaunchCleanupResponse> {
        let dispatch = dispatch_dir(store, run, false)?;
        let _lock = acquire_lock(store, run)?;
        let path = store
            .runs_root
            .join(run.as_str())
            .join("dispatch")
            .join(pending_name(launch_hash));
        let bytes = read_document(&path, "read pending dispatch")?;
        let mut pending: PendingDispatch = serde_json::from_slice(&bytes).map_err(|error| {
            DispatchStoreError::Domain(DispatchError::InvalidRecord(error.to_string()))
        })?;
        pending.validate()?;
        if pending.launch_state.is_terminal() {
            return Err(DispatchStoreError::Domain(
                DispatchError::PendingLaunchConsumed,
            ));
        }
        match state {
            PendingLaunchState::Expired => pending.expire()?,
            PendingLaunchState::Canceled => pending.cancel()?,
            PendingLaunchState::LaunchFailed => pending.fail()?,
            PendingLaunchState::Quarantined => {
                return Err(DispatchStoreError::Domain(DispatchError::InvalidPending(
                    "malignant quarantine is not a caller-selected cleanup state".into(),
                )));
            }
            PendingLaunchState::Pending
            | PendingLaunchState::ClaimedUnspawned
            | PendingLaunchState::Active => {
                return Err(DispatchStoreError::Domain(DispatchError::InvalidPending(
                    "launch cleanup target is not terminal".into(),
                )));
            }
        }
        safe_fs::replace_atomic(&path, &encode_document(&pending)?).map_err(|source| {
            DispatchStoreError::io("cleanup pending dispatch", path.clone(), source)
        })?;
        let _ = dispatch;
        let response = LaunchCleanupResponse {
            schema: shepherd::dispatch::LAUNCH_CLEANUP_SCHEMA.into(),
            launch_id_hash: launch_hash,
            state,
        };
        response.validate()?;
        Ok(response)
    }

    pub(super) fn load_root_binding(
        store: &DispatchStore,
        run: &RunId,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        dispatch_dir(store, run, false)?;
        let _lock = acquire_lock(store, run)?;
        ensure_no_root_renewal(store, session_id)?;
        let path = store.root_binding_path(run, session_id);
        let bytes = read_document(&path, "read root binding")?;
        decode_root_binding(&bytes, run, session_id)
    }

    pub(super) fn read_locked_root_binding(
        authority: &LockedDispatchRun<'_>,
        session_id: &SessionId,
    ) -> DispatchStoreResult<RootSessionBinding> {
        ensure_no_root_renewal(authority.store, session_id)?;
        let path = authority.dispatch.join(root_binding_name(session_id));
        decode_root_binding(
            &read_document(&path, "read root binding")?,
            authority.run,
            session_id,
        )
    }

    pub(super) fn read_locked_review_custody(
        authority: &LockedDispatchRun<'_>,
        subject: &AgentId,
    ) -> DispatchStoreResult<ReviewCustody> {
        let path = authority.dispatch.join(review_custody_name(subject));
        decode_review_custody(
            &read_document(&path, "read review custody")?,
            authority.run,
            subject,
        )
    }

    pub(super) fn read_locked_record(
        authority: &LockedDispatchRun<'_>,
        subject: &AgentId,
    ) -> DispatchStoreResult<DispatchRecord> {
        read_record(authority.store, authority.run, subject)
    }

    pub(super) fn read_locked_pending(
        authority: &LockedDispatchRun<'_>,
        launch_hash: [u8; 32],
    ) -> DispatchStoreResult<PendingDispatch> {
        read_pending(authority.store, authority.run, launch_hash)
    }

    pub(super) fn load_profile_lease(
        store: &DispatchStore,
        run: &RunId,
        session_id: &SessionId,
    ) -> DispatchStoreResult<ProfileLease> {
        dispatch_dir(store, run, false)?;
        let _lock = acquire_lock(store, run)?;
        let path = store.profile_lease_path(run, session_id);
        decode_profile_lease(
            &read_document(&path, "read profile lease")?,
            run,
            session_id,
        )
    }

    pub(super) fn load_review_custody(
        store: &DispatchStore,
        run: &RunId,
        subject: &AgentId,
    ) -> DispatchStoreResult<ReviewCustody> {
        dispatch_dir(store, run, false)?;
        let _lock = acquire_lock(store, run)?;
        let path = store.review_custody_path(run, subject);
        let custody =
            decode_review_custody(&read_document(&path, "read review custody")?, run, subject)?;
        if custody.state != ReviewCustodyState::Active {
            recover_malignant_subject(store, &custody, false)?;
        }
        Ok(custody)
    }

    pub(super) fn read_review_terminal_snapshot(
        store: &DispatchStore,
        run: &RunId,
        subject: &AgentId,
    ) -> DispatchStoreResult<ReviewTerminalSnapshot> {
        let dispatch = dispatch_dir(store, run, false)?;
        let _lock = acquire_lock(store, run)?;
        let custody = decode_review_custody(
            &read_document(
                &dispatch.join(review_custody_name(subject)),
                "read review custody",
            )?,
            run,
            subject,
        )?;
        ensure_no_root_renewal(store, &custody.root_session_id)?;
        Ok(ReviewTerminalSnapshot {
            root: decode_root_binding(
                &read_document(
                    &dispatch.join(root_binding_name(&custody.root_session_id)),
                    "read root binding",
                )?,
                run,
                &custody.root_session_id,
            )?,
            subject: read_record(store, run, subject)?,
            pending: read_pending(store, run, custody.pending_launch_id_hash)?,
            custody,
        })
    }

    pub(super) fn load_skill_use(
        store: &DispatchStore,
        run: &RunId,
        key: SkillUseKey<'_>,
        skill: &str,
    ) -> DispatchStoreResult<SkillUseChallenge> {
        dispatch_dir(store, run, false)?;
        let _lock = acquire_lock(store, run)?;
        validate_skill_use_root(store, run, key)?;
        let name = skill_use_name(key, skill)?;
        let path = store
            .runs_root
            .join(run.as_str())
            .join("dispatch")
            .join(name);
        decode_skill_use(
            &read_document(&path, "read skill-use challenge")?,
            run,
            key,
            skill,
        )
    }

    fn validate_skill_use_root(
        store: &DispatchStore,
        run: &RunId,
        key: SkillUseKey<'_>,
    ) -> DispatchStoreResult<()> {
        let SkillUseKey::Root(expected) = key else {
            return Ok(());
        };
        let session = &expected.binding.session_id;
        ensure_no_root_renewal(store, session)?;
        let binding = decode_root_binding(
            &read_document(&store.root_binding_path(run, session), "read root binding")?,
            run,
            session,
        )?;
        let current = decode_current_root_binding(
            &read_document(
                &store.current_root_binding_path(session),
                "read current root binding",
            )?,
            session,
        )?;
        let profile = match read_document(
            &store.profile_lease_path(run, session),
            "read profile lease",
        ) {
            Ok(bytes) => Some(decode_profile_lease(&bytes, run, session)?),
            Err(DispatchStoreError::Io { source, .. })
                if source.kind() == std::io::ErrorKind::NotFound =>
            {
                None
            }
            Err(error) => return Err(error),
        };
        if binding != expected.binding || current != binding || profile != expected.profile_lease {
            return Err(DispatchStoreError::Domain(DispatchError::InvalidSkillUse(
                "root or profile authority changed during the skill-use operation".into(),
            )));
        }
        Ok(())
    }

    pub(super) fn load_run(store: &DispatchStore, run: &RunId) -> DispatchStoreResult<RunState> {
        let path = store.runs_root.join(run.as_str()).join("run.json");
        let bytes = read_document(&path, "read run document")?;
        let state: RunState = serde_json::from_slice(&bytes).map_err(|error| {
            DispatchStoreError::InvalidRunDocument {
                path: path.clone(),
                reason: error.to_string(),
            }
        })?;
        if state.run != run.as_str()
            || state.schema_version != 1
            || !matches!(
                state.status.as_str(),
                "planted" | "planned" | "executing" | "closing" | "closed"
            )
        {
            return Err(DispatchStoreError::InvalidRunDocument {
                path,
                reason: "run identity, schema, or status is invalid".into(),
            });
        }
        Ok(state)
    }

    pub(super) fn load(
        store: &DispatchStore,
        run: &RunId,
        agent_id: &AgentId,
    ) -> DispatchStoreResult<DispatchRecord> {
        dispatch_dir(store, run, false).map_err(|error| {
            if is_not_found(&error) {
                unknown_record(run, agent_id, "dispatch directory is absent")
            } else {
                error
            }
        })?;
        let _lock = acquire_lock(store, run)?;
        let record = read_record(store, run, agent_id)?;
        let custody_path = store.review_custody_path(run, agent_id);
        match read_document(&custody_path, "read review custody")
            .and_then(|bytes| decode_review_custody(&bytes, run, agent_id))
        {
            Ok(custody) if custody.state != ReviewCustodyState::Active => {
                recover_malignant_subject(store, &custody, false).map(|(record, _)| record)
            }
            Ok(_) => Ok(record),
            Err(error) if is_not_found(&error) => Ok(record),
            Err(error) => Err(error),
        }
    }

    pub(super) fn read_artifact(
        store: &DispatchStore,
        run: &RunId,
        reference: &str,
    ) -> DispatchStoreResult<Vec<u8>> {
        let _ = dispatch_dir(store, run, false)?;
        let path = store.runs_root.join(run.as_str()).join(reference);
        safe_fs::read_regular_nofollow(&path, MAX_RECORD_BYTES).map_err(|source| {
            if source.kind() == std::io::ErrorKind::InvalidInput {
                DispatchStoreError::UnsafePath { path }
            } else {
                DispatchStoreError::io("read completion artifact", path, source)
            }
        })
    }

    pub(super) fn stop(
        store: &DispatchStore,
        run: &RunId,
        request: StopRequest,
    ) -> DispatchStoreResult<DispatchRecord> {
        dispatch_dir(store, run, false)?;
        let _lock = acquire_lock(store, run)?;
        let mut record = read_record(store, run, &request.agent_id)?;
        let mut request = request;
        request.stopped_at = current_time_millis();
        record.stop(request)?;
        record.validate_loaded()?;
        replace_record(store, run, &record)?;
        Ok(record)
    }

    pub(super) fn stop_verified(
        store: &DispatchStore,
        run: &RunId,
        native: &NativeIdentity,
        request: StopRequest,
    ) -> DispatchStoreResult<DispatchRecord> {
        dispatch_dir(store, run, false)?;
        let _lock = acquire_lock(store, run)?;
        let mut record = read_record(store, run, &request.agent_id)?;
        resolve_native_identity(Some(&record), native)?;
        let mut request = request;
        request.stopped_at = current_time_millis();
        record.stop(request)?;
        record.validate_loaded()?;
        replace_record(store, run, &record)?;
        Ok(record)
    }

    pub(super) fn quarantine_malignant(
        store: &DispatchStore,
        expected_custody: Option<&ReviewCustody>,
        expected: &DispatchRecord,
        expected_pending: &PendingDispatch,
        custody: &ReviewCustody,
    ) -> DispatchStoreResult<(DispatchRecord, PendingDispatch)> {
        dispatch_dir(store, &expected.run, false)?;
        let _lock = acquire_lock(store, &expected.run)?;
        let custody_path = store.review_custody_path(&expected.run, &expected.agent_id);
        let current_custody = read_document(&custody_path, "read review custody")
            .and_then(|bytes| decode_review_custody(&bytes, &expected.run, &expected.agent_id));
        let already_committed = current_custody
            .as_ref()
            .is_ok_and(|current| current == custody);
        if !already_committed {
            match (expected_custody, current_custody) {
                (Some(expected_custody), Ok(current)) if &current == expected_custody => {
                    safe_fs::replace_atomic(&custody_path, &encode_document(custody)?).map_err(
                        |source| {
                            DispatchStoreError::io(
                                "replace review custody",
                                custody_path.clone(),
                                source,
                            )
                        },
                    )?;
                }
                (None, Err(error)) if is_not_found(&error) => {
                    publish_no_clobber(&custody_path, &encode_document(custody)?)?;
                }
                _ => {
                    return Err(DispatchStoreError::Domain(
                        DispatchError::InvalidReviewCustody(
                            "review custody changed before subject quarantine".into(),
                        ),
                    ));
                }
            }
            if store.review_quarantine_fault == Some(ReviewQuarantineFault::AfterCustodyCommit) {
                return Err(DispatchStoreError::Reconciliation(
                    "injected failure after malignant custody commit".into(),
                ));
            }
        }
        let current_record = read_record(store, &expected.run, &expected.agent_id)?;
        let pending_path = store
            .runs_root
            .join(expected.run.as_str())
            .join("dispatch")
            .join(pending_name(expected_pending.launch_id_hash));
        let current_pending: PendingDispatch =
            serde_json::from_slice(&read_document(&pending_path, "read pending dispatch")?)
                .map_err(|error| {
                    DispatchStoreError::Domain(DispatchError::InvalidRecord(error.to_string()))
                })?;
        current_pending.validate()?;
        if current_record != *expected && current_record.state != DispatchState::Malignant {
            return Err(DispatchStoreError::Domain(
                DispatchError::InvalidReviewCustody(
                    "subject dispatch changed before malignant quarantine".into(),
                ),
            ));
        }
        if current_pending != *expected_pending
            && current_pending.launch_state != PendingLaunchState::Quarantined
        {
            return Err(DispatchStoreError::Domain(
                DispatchError::InvalidReviewCustody(
                    "subject pending claim changed before malignant quarantine".into(),
                ),
            ));
        }
        recover_malignant_subject(store, custody, !already_committed)
    }

    fn recover_malignant_subject(
        store: &DispatchStore,
        custody: &ReviewCustody,
        inject_fault: bool,
    ) -> DispatchStoreResult<(DispatchRecord, PendingDispatch)> {
        let stopped_at = custody.stopped_at.ok_or_else(|| {
            DispatchStoreError::Domain(DispatchError::InvalidReviewCustody(
                "terminal review custody has no stop time".into(),
            ))
        })?;
        let mut record = read_record(store, &custody.run, &custody.subject_agent_id)?;
        if record.root_session_id != custody.root_session_id
            || record.session_id != custody.subject_session_id
            || record.role != custody.subject_role
            || record.lane != custody.lane
        {
            return Err(DispatchStoreError::Domain(
                DispatchError::InvalidReviewCustody(
                    "terminal custody does not match the durable dispatch identity".into(),
                ),
            ));
        }
        match record.state {
            DispatchState::Active => {
                record.quarantine_malignant(stopped_at)?;
                replace_record(store, &custody.run, &record)?;
            }
            DispatchState::Malignant if record.stopped_at == Some(stopped_at) => {}
            _ => {
                return Err(DispatchStoreError::Domain(
                    DispatchError::InvalidReviewCustody(
                        "terminal custody conflicts with the durable dispatch state".into(),
                    ),
                ));
            }
        }
        if inject_fault
            && store.review_quarantine_fault == Some(ReviewQuarantineFault::AfterRecordCommit)
        {
            return Err(DispatchStoreError::Reconciliation(
                "injected failure after malignant dispatch commit".into(),
            ));
        }
        let pending_path = store
            .runs_root
            .join(custody.run.as_str())
            .join("dispatch")
            .join(pending_name(custody.pending_launch_id_hash));
        let mut pending: PendingDispatch =
            serde_json::from_slice(&read_document(&pending_path, "read pending dispatch")?)
                .map_err(|error| {
                    DispatchStoreError::Domain(DispatchError::InvalidRecord(error.to_string()))
                })?;
        pending.validate()?;
        if pending.root_session_id != custody.root_session_id
            || pending.expected_attachment.agent_id != custody.subject_agent_id
            || pending.expected_child_session_id != custody.subject_session_id
        {
            return Err(DispatchStoreError::Domain(
                DispatchError::InvalidReviewCustody(
                    "terminal custody does not match the durable pending claim".into(),
                ),
            ));
        }
        match pending.launch_state {
            PendingLaunchState::Active => {
                pending.quarantine()?;
                safe_fs::replace_atomic(&pending_path, &encode_document(&pending)?).map_err(
                    |source| {
                        DispatchStoreError::io(
                            "quarantine pending dispatch",
                            pending_path.clone(),
                            source,
                        )
                    },
                )?;
            }
            PendingLaunchState::Quarantined => {}
            _ => {
                return Err(DispatchStoreError::Domain(
                    DispatchError::InvalidReviewCustody(
                        "terminal custody conflicts with the durable pending claim".into(),
                    ),
                ));
            }
        }
        Ok((record, pending))
    }

    /// Resolve `<runs_root>/<run>/dispatch`, creating it only when asked.
    ///
    /// `create` is false on every read path, so a caller expecting an existing
    /// ledger gets `NotFound` instead of quietly manufacturing an empty
    /// directory that then reports "record is absent".
    fn dispatch_dir(
        store: &DispatchStore,
        run: &RunId,
        create: bool,
    ) -> DispatchStoreResult<PathBuf> {
        reject_parent_components(&store.runs_root)?;
        let dispatch = store.runs_root.join(run.as_str()).join("dispatch");
        if create {
            std::fs::create_dir_all(&dispatch).map_err(|source| {
                DispatchStoreError::io("create dispatch directory", dispatch.clone(), source)
            })?;
        }
        safe_fs::reject_link_components(&dispatch).map_err(|_| DispatchStoreError::UnsafePath {
            path: dispatch.clone(),
        })?;
        if !create {
            let metadata = std::fs::symlink_metadata(&dispatch).map_err(|source| {
                DispatchStoreError::io("open dispatch directory", dispatch.clone(), source)
            })?;
            if !metadata.is_dir() {
                return Err(DispatchStoreError::UnsafePath { path: dispatch });
            }
        }
        Ok(dispatch)
    }

    /// Read `<runs_root>/<run>/run.json`.
    ///
    /// The anchor carries nothing, so it is named `_root`: every open below
    /// re-walks and re-checks its own components (`crate::safe_fs`), which is
    /// exactly the guarantee the module header refuses to overstate. An absent
    /// document surfaces as `Io { NotFound }` and a linked or non-regular one
    /// as `UnsafePath`, so the shared resolver can tell "not a run" apart from
    /// "unreadable".
    pub(super) fn read_run_document(
        store: &DispatchStore,
        _root: &RunsRoot,
        run: &RunId,
    ) -> DispatchStoreResult<RunState> {
        let path = store.runs_root.join(run.as_str()).join("run.json");
        let bytes = read_document(&path, "read run document")?;
        let state: RunState = serde_json::from_slice(&bytes).map_err(|error| {
            DispatchStoreError::InvalidRunDocument {
                path: path.clone(),
                reason: error.to_string(),
            }
        })?;
        if state.run != run.as_str()
            || state.schema_version != 1
            || !matches!(
                state.status.as_str(),
                "planted" | "planned" | "executing" | "closing" | "closed"
            )
        {
            return Err(DispatchStoreError::InvalidRunDocument {
                path,
                reason: "run identity, schema, or status is invalid".into(),
            });
        }
        Ok(state)
    }

    fn read_record(
        store: &DispatchStore,
        run: &RunId,
        agent_id: &AgentId,
    ) -> DispatchStoreResult<DispatchRecord> {
        let path = store.record_path(run, agent_id);
        let bytes = match read_document(&path, "read dispatch record") {
            Ok(bytes) => bytes,
            Err(error) if is_not_found(&error) => {
                return Err(unknown_record(run, agent_id, "record is absent"));
            }
            Err(error @ DispatchStoreError::UnsafePath { .. }) => return Err(error),
            Err(error) => return Err(unknown_record(run, agent_id, error.to_string())),
        };
        let record: DispatchRecord = serde_json::from_slice(&bytes)
            .map_err(|error| unknown_record(run, agent_id, error.to_string()))?;
        record
            .validate_loaded()
            .map_err(|error| unknown_record(run, agent_id, error.to_string()))?;
        if &record.agent_id != agent_id || &record.run != run {
            return Err(unknown_record(
                run,
                agent_id,
                "record identity does not match its canonical path",
            ));
        }
        Ok(record)
    }

    fn read_document(path: &Path, operation: &'static str) -> DispatchStoreResult<Vec<u8>> {
        safe_fs::read_regular_nofollow(path, MAX_RECORD_BYTES).map_err(|source| {
            if source.kind() == std::io::ErrorKind::InvalidInput {
                DispatchStoreError::UnsafePath {
                    path: path.to_path_buf(),
                }
            } else {
                DispatchStoreError::io(operation, path.to_path_buf(), source)
            }
        })
    }

    /// The advisory lock is `std::fs::File::try_lock`: `LockFileEx` on Windows,
    /// `flock` on unix. The unix twin makes the same call, so the
    /// mutual-exclusion contract is genuinely shared, not approximated.
    fn acquire_lock(store: &DispatchStore, run: &RunId) -> DispatchStoreResult<DispatchLock> {
        let path = store.runs_root.join(run.as_str()).join("run.lock");
        let lock_path = path.clone();
        if safe_fs::is_link(&lock_path).unwrap_or(true) {
            return Err(DispatchStoreError::UnsafePath { path: lock_path });
        }
        let file = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&lock_path)
            .map_err(|source| DispatchStoreError::io("open dispatch lock", path.clone(), source))?;
        let started = Instant::now();
        loop {
            match file.try_lock() {
                Ok(()) => return Ok(DispatchLock(file)),
                Err(TryLockError::WouldBlock) if started.elapsed() < store.timeout => {
                    let remaining = store.timeout.saturating_sub(started.elapsed());
                    std::thread::sleep(LOCK_RETRY_INTERVAL.min(remaining));
                }
                Err(TryLockError::WouldBlock) => {
                    return Err(DispatchStoreError::LockTimeout {
                        path,
                        timeout: store.timeout,
                    });
                }
                Err(TryLockError::Error(source)) => {
                    return Err(DispatchStoreError::io(
                        "acquire dispatch lock",
                        path,
                        source,
                    ));
                }
            }
        }
    }

    fn acquire_current_root_lock(store: &DispatchStore) -> DispatchStoreResult<DispatchLock> {
        let path = store.runs_root.join(".root-session.lock");
        if safe_fs::is_link(&path).unwrap_or(true) {
            return Err(DispatchStoreError::UnsafePath { path });
        }
        let file = std::fs::OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(&path)
            .map_err(|source| {
                DispatchStoreError::io("open current root lock", path.clone(), source)
            })?;
        let started = Instant::now();
        loop {
            match file.try_lock() {
                Ok(()) => return Ok(DispatchLock(file)),
                Err(TryLockError::WouldBlock) if started.elapsed() < store.timeout => {
                    let remaining = store.timeout.saturating_sub(started.elapsed());
                    std::thread::sleep(LOCK_RETRY_INTERVAL.min(remaining));
                }
                Err(TryLockError::WouldBlock) => {
                    return Err(DispatchStoreError::LockTimeout {
                        path,
                        timeout: store.timeout,
                    });
                }
                Err(TryLockError::Error(source)) => {
                    return Err(DispatchStoreError::io(
                        "acquire current root lock",
                        path,
                        source,
                    ));
                }
            }
        }
    }

    fn read_optional_path(path: &Path) -> DispatchStoreResult<Option<Vec<u8>>> {
        match safe_fs::read_regular_nofollow(path, MAX_RECORD_BYTES) {
            Ok(bytes) => Ok(Some(bytes)),
            Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(None),
            Err(source) => Err(DispatchStoreError::io(
                "read singleton publication",
                path.to_path_buf(),
                source,
            )),
        }
    }

    fn publish_final_no_clobber(
        store: &DispatchStore,
        run: &RunId,
        temp: &Path,
        target: &Path,
        bytes: &[u8],
    ) -> DispatchStoreResult<()> {
        if let Some(existing) = read_optional_path(target)? {
            if existing != bytes {
                return Err(DispatchStoreError::SingletonPublication {
                    nonce: temp.display().to_string(),
                    reason: "final dispatch name contains different bytes".into(),
                });
            }
            safe_fs::remove_file_nofollow(temp).map_err(|source| {
                DispatchStoreError::io(
                    "remove singleton preparing file",
                    temp.to_path_buf(),
                    source,
                )
            })?;
            return Ok(());
        }
        match safe_fs::write_no_clobber(target, bytes) {
            Ok(true) => {}
            Ok(false) => {
                let existing = read_optional_path(target)?;
                if existing.as_deref() != Some(bytes) {
                    return Err(DispatchStoreError::SingletonPublication {
                        nonce: temp.display().to_string(),
                        reason: "racing final dispatch publication differs".into(),
                    });
                }
            }
            Err(source) => {
                return Err(DispatchStoreError::io(
                    "publish singleton dispatch record",
                    target.to_path_buf(),
                    source,
                ));
            }
        }
        safe_fs::remove_file_nofollow(temp).map_err(|source| {
            DispatchStoreError::io(
                "remove singleton preparing file",
                temp.to_path_buf(),
                source,
            )
        })?;
        let _ = store;
        let _ = run;
        Ok(())
    }

    /// Refusing to replace is the point: a second publication for a live
    /// binding must be reported, never silently overwritten.
    fn publish_no_clobber(path: &Path, bytes: &[u8]) -> DispatchStoreResult<()> {
        match safe_fs::write_no_clobber(path, bytes) {
            Ok(true) => Ok(()),
            Ok(false) => Err(DispatchStoreError::AlreadyExists {
                path: path.to_path_buf(),
            }),
            Err(source) if source.kind() == std::io::ErrorKind::InvalidInput => {
                Err(DispatchStoreError::UnsafePath {
                    path: path.to_path_buf(),
                })
            }
            Err(source) => Err(DispatchStoreError::io(
                "publish dispatch record",
                path.to_path_buf(),
                source,
            )),
        }
    }

    fn remove_record(
        store: &DispatchStore,
        run: &RunId,
        agent_id: &AgentId,
    ) -> DispatchStoreResult<()> {
        let path = store.record_path(run, agent_id);
        let metadata = match std::fs::symlink_metadata(&path) {
            Ok(metadata) => metadata,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
            Err(error) => {
                return Err(DispatchStoreError::io(
                    "inspect active record",
                    path.clone(),
                    error,
                ));
            }
        };
        if !metadata.is_file() || metadata.file_type().is_symlink() {
            return Err(DispatchStoreError::UnsafePath { path });
        }
        safe_fs::remove_file_nofollow(&path)
            .map_err(|source| DispatchStoreError::io("remove orphaned active record", path, source))
    }

    fn replace_record(
        store: &DispatchStore,
        run: &RunId,
        record: &DispatchRecord,
    ) -> DispatchStoreResult<()> {
        let path = store.record_path(run, &record.agent_id);
        let bytes = encode_record(record)?;
        safe_fs::replace_atomic(&path, &bytes).map_err(|source| {
            DispatchStoreError::io("replace dispatch record", path.clone(), source)
        })
    }

    fn encode_record(record: &DispatchRecord) -> DispatchStoreResult<Vec<u8>> {
        encode_document(record)
    }

    fn encode_document(document: &impl serde::Serialize) -> DispatchStoreResult<Vec<u8>> {
        let mut bytes = serde_json::to_vec(document).map_err(|error| {
            DispatchStoreError::Domain(DispatchError::InvalidRecord(error.to_string()))
        })?;
        bytes.push(b'\n');
        let max = usize::try_from(MAX_RECORD_BYTES).expect("record limit fits usize");
        if bytes.len() > max {
            return Err(DispatchStoreError::RecordTooLarge {
                size: bytes.len(),
                max,
            });
        }
        Ok(bytes)
    }

    fn record_name(agent_id: &AgentId) -> String {
        format!("{}.json", agent_id.as_str())
    }

    fn unknown_record(
        run: &RunId,
        agent_id: &AgentId,
        reason: impl Into<String>,
    ) -> DispatchStoreError {
        DispatchStoreError::UnknownRecord {
            run: run.clone(),
            agent_id: agent_id.clone(),
            reason: reason.into(),
        }
    }

    fn reject_parent_components(path: &Path) -> DispatchStoreResult<()> {
        if !path.is_absolute()
            || path.components().any(|component| {
                matches!(
                    component,
                    std::path::Component::ParentDir | std::path::Component::CurDir
                )
            })
        {
            return Err(DispatchStoreError::UnsafePath {
                path: path.to_path_buf(),
            });
        }
        Ok(())
    }

    struct DispatchLock(File);

    impl Drop for DispatchLock {
        fn drop(&mut self) {
            let _ = self.0.unlock();
        }
    }
}