shepherd-cli 6.5.9

//! Native request boundary over the pure dispatch engine and primary-run store.

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

#[cfg(unix)]
use std::fs::File;
#[cfg(unix)]
use std::io::Read;

use sha2::{Digest, Sha256};

use crate::portable_path::portable_absolute_path;
use shepherd::{
    Harness,
    dispatch::{
        AgentId, AgentType, AttachmentKind, CapabilityProbe, CapabilityReadiness, CapabilityReport,
        CarrierAttachmentExpectation, DispatchError, DispatchId, DispatchRecord, DispatchStart,
        DispatchState, GitCommit, IdentityError, IdentityResolution, LaneId,
        LoadedCarrierAttestationV1, LoadedSkillAttestation, NativeIdentity, PathAuthority,
        PathClass, PathFactKind, PendingDispatch, PendingLaunchState, Profile,
        ProfileAttachmentExpectation, ProfileLease, ProfileLeaseState, ProjectFilesystemId,
        ProjectId, ROOT_SESSION_SCHEMA, ReviewCustody, ReviewCustodyState, ReviewResult,
        ReviewRuling, Role, RootSessionBinding, RunId, SessionId, SkillUseChallenge,
        SkillUsePrepare, SkillUseRootAuthority, SkillUseStage, SkillUseState, StartupAttachment,
        StopRequest, TrustedPathFacts, WorkKind, classify_trusted_path, constant_time_digest_eq,
        path_in_write_scope, validate_pending_edge,
    },
    registry::{
        DispatchSingletonInput, DispatchSingletonPublication, DispatchSingletonPublicationInput,
        Error as RegistryError, Registry,
    },
};

use crate::{
    DispatchStore, DispatchStoreError, DispatchStoreResult, dispatch_broker::BrokerLaunchId,
    dispatch_scope::derive_write_paths,
};

const REQUEST_SCHEMA: &str = "shepherd.dispatch-request/1";
pub const REVIEW_REQUEST_SCHEMA: &str = "shepherd.review-request/1";
pub const REVIEW_RULING_REQUEST_SCHEMA: &str = "shepherd.review-ruling-request/1";
pub const REVIEW_REPLACEMENT_REQUEST_SCHEMA: &str = "shepherd.review-replacement-request/1";
const PENDING_REQUEST_SCHEMA: &str = "shepherd.pending-dispatch-request/2";
const MAX_LEASE_MS: u64 = 86_400_000;
const MAX_PENDING_BYTES: usize = 1_048_576;
const MAX_ATTESTATION_BYTES: usize = 256 * 1024;
const MAX_NATIVE_BINARY_BYTES: usize = 256 * 1024 * 1024;
const MAX_ATTACHMENT_FILES: usize = 4_096;
const MAX_ATTACHMENT_TREE_BYTES: usize = 8 * 1024 * 1024;
const MAX_ATTACHMENT_DEPTH: usize = 32;
const PROFILE_REQUEST_SCHEMA: &str = "shepherd.profile-request/1";
const SKILL_USE_REQUEST_SCHEMA: &str = "shepherd.skill-use-request/1";

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct ReviewValidationRequest {
    pub schema: String,
    pub review: ReviewResult,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct ReviewRulingRequest {
    pub schema: String,
    pub run: String,
    pub harness: Harness,
    pub root_session_id: String,
    pub subject_agent_id: String,
    pub reviewer_dispatch_id: String,
    pub reviewer_session_id: String,
    pub task_generation: u32,
    pub review: ReviewResult,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct ReviewReplacementRequest {
    pub schema: String,
    pub run: String,
    pub harness: Harness,
    pub root_session_id: String,
    pub subject_agent_id: String,
    pub replacement_agent_id: String,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct ReviewTerminalVerificationRequest {
    schema: String,
    run: String,
    harness: Harness,
    root_session_id: String,
    subject_agent_id: String,
    subject_session_id: String,
    task_generation: u32,
    task_sha256: String,
    pending_launch_id_hash: String,
}

pub fn validate_review_result_request(
    request: ReviewValidationRequest,
) -> DispatchServiceResult<ReviewResult> {
    if request.schema != REVIEW_REQUEST_SCHEMA {
        return Err(DispatchServiceError::InvalidRequest(format!(
            "unsupported review schema `{}`",
            request.schema
        )));
    }
    request.review.validate()?;
    Ok(request.review)
}

fn review_document_sha256(value: &impl serde::Serialize) -> DispatchServiceResult<String> {
    let mut document = serde_json::to_value(value).map_err(|error| {
        DispatchServiceError::InvalidRequest(format!("cannot encode review proof: {error}"))
    })?;
    document.sort_all_objects();
    let bytes = serde_json::to_vec(&document).map_err(|error| {
        DispatchServiceError::InvalidRequest(format!("cannot encode review proof: {error}"))
    })?;
    Ok(hex(&sha256(&bytes)))
}

fn validate_review_claim_custody(custody: &ReviewCustody) -> Result<(), DispatchError> {
    custody.validate()?;
    if custody.state != ReviewCustodyState::Active {
        return Err(DispatchError::ReviewCustodyTerminal);
    }
    Ok(())
}

fn review_resume_input(subject: &DispatchRecord, now: i64) -> DispatchServiceResult<DispatchStart> {
    let probe = &subject.capabilities;
    Ok(DispatchStart {
        project_id: subject.project_id.clone(),
        run: subject.run.clone(),
        root_session_id: subject.root_session_id.clone(),
        run_incarnation: subject.run_incarnation.clone(),
        nonce: subject.nonce.clone(),
        harness: subject.harness,
        // An unpublished argument to a pure domain predicate, never a grant.
        agent_id: AgentId::new(format!(
            "review-preflight-{}",
            hex(&sha256(subject.agent_id.as_str().as_bytes()))
        ))?,
        agent_type: subject.agent_type.clone(),
        role: subject.role,
        lane: subject.lane.clone(),
        parent_agent_id: subject.parent_agent_id.clone(),
        session_id: subject.session_id.clone(),
        write_scope: subject.write_scope.clone(),
        model: subject.model.clone(),
        capability_contract: subject.role.dispatch_capability_contract()?,
        capability_probe: CapabilityProbe {
            probe_id: probe.probe_id.clone(),
            observed: probe.observed.clone(),
            observed_events: probe.observed_events.clone(),
            source: probe.source.clone(),
            harness_version: probe.harness_version.clone(),
            provider_version: probe.provider_version.clone(),
            binary_sha256: probe.binary_sha256.clone(),
            package_sha256: probe.package_sha256.clone(),
            probed_at: probe.probed_at,
        },
        startup_attachment: subject.startup_attachment.clone(),
        attachment_nonce: subject.attachment_nonce.clone(),
        result_artifact: subject.result_artifact.clone(),
        result_nonce: subject.result_nonce.clone(),
        review_artifact: subject.review_artifact.clone(),
        review_nonce: subject.review_nonce.clone(),
        started_at: now,
        lease_expires_at: subject.lease_expires_at,
        resumes_agent_id: Some(subject.agent_id.clone()),
    })
}

pub type DispatchServiceResult<T> = core::result::Result<T, DispatchServiceError>;

#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum DispatchServiceError {
    #[error("invalid dispatch request: {0}")]
    InvalidRequest(String),
    #[error(transparent)]
    Store(#[from] DispatchStoreError),
    #[error(transparent)]
    Domain(#[from] DispatchError),
    #[error(transparent)]
    Identity(#[from] IdentityError),
    #[error(
        "root session `{session_id}` is already bound to run `{run}` with a different identity"
    )]
    RootBindingConflict { run: RunId, session_id: SessionId },
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct BindRootDispatchRequest {
    pub schema: String,
    #[serde(default)]
    pub run: Option<String>,
    pub harness: Harness,
    pub session_id: String,
    pub role_carrier: String,
    pub mode: String,
    pub lease_ms: u64,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct ResolveDispatchRequest {
    pub schema: String,
    #[serde(default)]
    pub run: Option<String>,
    pub harness: Harness,
    pub agent_id: Option<String>,
    pub agent_type: Option<String>,
    pub role_carrier: Option<String>,
    pub lane: Option<String>,
    pub session_id: String,
    pub tool_call_id: Option<String>,
    #[serde(default)]
    pub tool_name: Option<String>,
    #[serde(default)]
    pub tool_input: Option<serde_json::Value>,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct StopDispatchRequest {
    pub schema: String,
    #[serde(default)]
    pub run: Option<String>,
    pub harness: Harness,
    pub agent_id: String,
    pub agent_type: String,
    pub role_carrier: Option<String>,
    pub lane: Option<String>,
    pub session_id: String,
    pub expected_revision: u64,
    pub result_artifact: Option<String>,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct CarrierAttachmentExpectationRequest {
    pub target: Harness,
    pub role: String,
    pub agent_id: String,
    pub attachment_kind: String,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfileEnterRequest {
    pub schema: String,
    pub run: String,
    pub harness: Harness,
    pub session_id: String,
    pub lease_ms: u64,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfileActivateRequest {
    pub schema: String,
    pub run: String,
    pub harness: Harness,
    pub session_id: String,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct ProfileExitRequest {
    pub schema: String,
    pub run: String,
    pub harness: Harness,
    pub session_id: String,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct SkillUsePrepareRequest {
    pub schema: String,
    pub run: String,
    pub harness: Harness,
    pub dispatch_id: Option<String>,
    pub session_id: String,
    pub skill: String,
    pub stage: SkillUseStage,
    pub lease_ms: u64,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct SkillUseAttestRequest {
    pub schema: String,
    pub run: String,
    pub harness: Harness,
    pub dispatch_id: Option<String>,
    pub session_id: String,
    pub skill: String,
}

#[derive(Debug, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct NativeTransportDescriptor {
    schema: String,
    binary: String,
    project_root: String,
    installed_package_root: String,
    installed_manifest: String,
    candidate_sha256: String,
    installed_manifest_sha256: String,
    env: serde_json::Value,
    #[serde(default)]
    auth_snapshot: Option<String>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct InstalledPackageBinding {
    binary: PathBuf,
    root: PathBuf,
    manifest: PathBuf,
    root_filesystem_id: ProjectFilesystemId,
    candidate_sha256: [u8; 32],
    manifest_sha256: [u8; 32],
}

struct TrustedSkillCarrier {
    installed_carrier_path: String,
    candidate_sha256: [u8; 32],
    carrier_sha256: [u8; 32],
    compiler_tree_sha256: [u8; 32],
    startup_skill: String,
    skill_bundle_sha256: [u8; 32],
}

#[derive(Eq, PartialEq)]
enum SkillUseAuthority {
    Child(Box<DispatchRecord>),
    Root(Box<SkillUseRootAuthority>),
}

impl SkillUseAuthority {
    fn role(&self) -> Role {
        match self {
            Self::Child(record) => record.role,
            Self::Root(authority) => authority.role(),
        }
    }

    fn root(&self) -> Option<&SkillUseRootAuthority> {
        match self {
            Self::Child(_) => None,
            Self::Root(authority) => Some(authority),
        }
    }

    fn expires_at(&self) -> i64 {
        match self {
            Self::Child(record) => record.lease_expires_at,
            Self::Root(authority) => authority
                .profile_lease
                .as_ref()
                .filter(|lease| lease.state == ProfileLeaseState::Active)
                .map_or(authority.binding.expires_at, |lease| {
                    lease.expires_at.min(authority.binding.expires_at)
                }),
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct PreparePendingDispatchRequest {
    pub schema: String,
    #[serde(default)]
    pub run: Option<String>,
    pub role: String,
    pub work_kind: String,
    pub lane: Option<String>,
    pub parent_dispatch_id: Option<String>,
    #[serde(default)]
    pub replaces_agent_id: Option<String>,
    pub baseline: String,
    pub read_scope: Vec<String>,
    pub write_scope: Vec<String>,
    pub result_artifact: String,
    pub review_artifact: String,
    pub task_file: String,
    pub child_session_id: String,
    pub lease_ms: u64,
    pub expected_attachment: CarrierAttachmentExpectationRequest,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ClaimPendingDispatchRequest {
    pub schema: String,
    pub launch_id: BrokerLaunchId,
    pub agent_id: String,
    pub session_id: String,
    pub agent_type: String,
    pub attestation: LoadedCarrierAttestationV1,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PreparePendingDispatchResponse {
    pub schema: String,
    pub launch_id: BrokerLaunchId,
    pub pending: PendingDispatch,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ClaimPendingDispatchResponse {
    pub schema: String,
    pub launch_id: BrokerLaunchId,
    pub pending: PendingDispatch,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct DispatchResolution {
    pub schema: String,
    pub project_id: ProjectId,
    pub run: RunId,
    pub harness: Harness,
    pub agent_id: Option<String>,
    pub agent_type: Option<AgentType>,
    pub role: Role,
    pub lane: Option<LaneId>,
    pub session_id: SessionId,
    pub write_scope: Vec<String>,
    pub capabilities: Option<CapabilityReport>,
    pub tool_call_id: Option<String>,
    pub mode: Option<String>,
    pub write_paths: Vec<String>,
    pub path_in_write_scope: Option<bool>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct DispatchService {
    store: DispatchStore,
    project_id: ProjectId,
    /// Source checkout/worktree where the authenticated process may act.
    /// The store and registry paths remain independently primary-custodied.
    workspace_root: Option<PathBuf>,
    registry_path: Option<PathBuf>,
    /// Native-installed compiler output. This is deliberately independent of
    /// the project checkout so project-controlled lookalikes cannot attest a
    /// carrier or on-demand skill.
    installed_package: Option<InstalledPackageBinding>,
    installed_package_error: Option<String>,
}

impl DispatchService {
    pub fn new(store: DispatchStore, project_id: ProjectId) -> Self {
        Self {
            store,
            project_id,
            workspace_root: None,
            registry_path: None,
            installed_package: None,
            installed_package_error: None,
        }
    }

    pub fn with_project_root(
        store: DispatchStore,
        project_id: ProjectId,
        project_root: impl Into<PathBuf>,
    ) -> Self {
        let workspace_root = project_root.into();
        let manifest = workspace_root.join(".shepherd-generated.json");
        let installed = local_installed_package_binding(&workspace_root, &manifest);
        Self {
            store,
            project_id,
            registry_path: Some(workspace_root.join(".shepherd/shepherd.db")),
            installed_package: installed.as_ref().ok().cloned(),
            installed_package_error: installed.err().map(|error| error.to_string()),
            workspace_root: Some(workspace_root),
        }
    }

    pub fn with_context(
        store: DispatchStore,
        project_id: ProjectId,
        workspace_root: impl Into<PathBuf>,
        registry_path: impl Into<PathBuf>,
    ) -> Self {
        let workspace_root = workspace_root.into();
        let installed = installed_package_binding_from_environment(&workspace_root);
        Self {
            store,
            project_id,
            workspace_root: Some(workspace_root),
            registry_path: Some(registry_path.into()),
            installed_package: installed.as_ref().ok().cloned(),
            installed_package_error: installed.err().map(|error| error.to_string()),
        }
    }

    /// Test and workspace bootstrap injection for one already-retained native
    /// package descriptor. Production `with_context` obtains the same values
    /// only from `SHEPHERD_NATIVE_DESCRIPTOR`.
    pub fn with_installed_package(
        mut self,
        root: impl Into<PathBuf>,
        manifest: impl Into<PathBuf>,
    ) -> Self {
        let root = root.into();
        let manifest = manifest.into();
        match local_installed_package_binding(&root, &manifest) {
            Ok(binding) => {
                self.installed_package = Some(binding);
                self.installed_package_error = None;
            }
            Err(error) => {
                self.installed_package = None;
                self.installed_package_error = Some(error.to_string());
            }
        }
        self
    }

    fn installed_package(&self) -> DispatchServiceResult<&InstalledPackageBinding> {
        self.installed_package.as_ref().ok_or_else(|| {
            DispatchServiceError::InvalidRequest(
                self.installed_package_error
                    .clone()
                    .unwrap_or_else(|| "native installed-package authority is absent".into()),
            )
        })
    }

    fn validate_current_root_binding(
        &self,
        binding: &RootSessionBinding,
    ) -> DispatchServiceResult<()> {
        let current = self.store.load_current_root_binding(&binding.session_id)?;
        if current != *binding {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "root session authority moved from run `{}` to run `{}`",
                binding.run, current.run
            )));
        }
        Ok(())
    }

    fn validate_dispatch_root_binding(
        &self,
        record: &DispatchRecord,
        now: i64,
    ) -> DispatchServiceResult<()> {
        let root = self
            .store
            .load_root_binding_for_run(&record.run, &record.root_session_id)?;
        self.validate_current_root_binding(&root)?;
        if root.project_id != record.project_id
            || root.harness != record.harness
            || now < root.bound_at
            || now >= root.expires_at
        {
            return Err(DispatchServiceError::InvalidRequest(
                "dispatch root binding is stale or does not own the child record".into(),
            ));
        }
        match self
            .store
            .load_review_custody(&record.run, &record.agent_id)
        {
            Ok(custody) if custody.state != ReviewCustodyState::Active => {
                return Err(DispatchServiceError::Domain(
                    DispatchError::ReviewCustodyTerminal,
                ));
            }
            Ok(_) => {}
            Err(DispatchStoreError::Io { source, .. })
                if source.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(error.into()),
        }
        Ok(())
    }

    #[must_use]
    pub fn store(&self) -> &DispatchStore {
        &self.store
    }

    fn prepare_singleton_publication(
        &self,
        record: &DispatchRecord,
        prepared_at: i64,
        replacement: Option<&crate::dispatch_store::ReviewTerminalSnapshot>,
    ) -> DispatchStoreResult<Option<(Registry, DispatchSingletonPublication)>> {
        if !matches!(record.role, Role::Engineer | Role::Conductor) {
            return Ok(None);
        }
        let registry_path = self.registry_path.as_ref().ok_or_else(|| {
            DispatchStoreError::Reconciliation(
                "singleton activation requires the authoritative project registry".into(),
            )
        })?;
        let mut bytes = serde_json::to_vec(record).map_err(|error| {
            DispatchStoreError::Domain(DispatchError::InvalidRecord(error.to_string()))
        })?;
        bytes.push(b'\n');
        let record_json = String::from_utf8(bytes.clone()).map_err(|error| {
            DispatchStoreError::Domain(DispatchError::InvalidRecord(error.to_string()))
        })?;
        let claim = DispatchSingletonInput {
            project_id: record.project_id.to_string(),
            run_id: record.run.to_string(),
            role: record.role.to_string(),
            lane_id: record.lane.as_ref().map(ToString::to_string),
            agent_id: record.agent_id.to_string(),
            harness: record.harness.to_string(),
            agent_type: record.agent_type.to_string(),
            parent_agent_id: record.parent_agent_id.as_ref().map(ToString::to_string),
            session_id: record.session_id.to_string(),
            write_scope: record.write_scope.clone(),
            claimed_at: prepared_at,
            resumes_agent_id: record.resumes_agent_id.as_ref().map(ToString::to_string),
        };
        let input = DispatchSingletonPublicationInput {
            nonce: record.nonce.clone(),
            record_path: format!("{}/dispatch/{}.json", record.run, record.agent_id),
            record_sha256: hex(&sha256(&bytes)),
            record_json,
            claim,
            prepared_at,
        };
        let mut registry = Registry::open_migrated(registry_path)
            .map_err(|error| DispatchStoreError::Reconciliation(error.to_string()))?;
        let publication = registry
            .transaction_immediate::<_, RegistryError, _>(|transaction| {
                if let Some(source) = replacement {
                    transaction.prepare_review_replacement_singleton(
                        &input,
                        &source.subject,
                        &source.pending,
                        &source.custody,
                    )
                } else {
                    transaction.prepare_dispatch_singleton(&input)
                }
            })
            .map_err(|error| DispatchStoreError::Reconciliation(error.to_string()))?;
        Ok(Some((registry, publication)))
    }

    fn refresh_singleton_publication(
        &self,
        record: &DispatchRecord,
        updated_at: i64,
    ) -> DispatchServiceResult<()> {
        if !matches!(record.role, Role::Engineer | Role::Conductor) {
            return Ok(());
        }
        let Some(registry_path) = self.registry_path.as_ref() else {
            return Ok(());
        };
        let mut registry = Registry::open_migrated(registry_path)
            .map_err(|error| DispatchStoreError::Reconciliation(error.to_string()))?;
        let lane_key = match record.role {
            Role::Engineer => "__run__".to_owned(),
            Role::Conductor => record
                .lane
                .as_ref()
                .ok_or_else(|| {
                    DispatchError::InvalidRecord("Conductor singleton has no lane".into())
                })?
                .to_string(),
            _ => unreachable!(),
        };
        let Some(claim) = registry
            .load_dispatch_singleton(
                record.project_id.as_str(),
                record.run.as_str(),
                record.role.as_str(),
                &lane_key,
            )
            .map_err(|error| DispatchStoreError::Reconciliation(error.to_string()))?
        else {
            return Ok(());
        };
        if claim.agent_id != record.agent_id.as_str() {
            return Err(DispatchError::InvalidRecord(
                "terminal singleton record does not match its current claim".into(),
            )
            .into());
        }
        let nonce = claim.publication_nonce.ok_or_else(|| {
            DispatchStoreError::Reconciliation(
                "current singleton claim has no publication nonce".into(),
            )
        })?;
        let mut bytes = serde_json::to_vec(record)
            .map_err(|error| DispatchError::InvalidRecord(error.to_string()))?;
        bytes.push(b'\n');
        let record_json = String::from_utf8(bytes.clone())
            .map_err(|error| DispatchError::InvalidRecord(error.to_string()))?;
        let digest = hex(&sha256(&bytes));
        registry
            .transaction_immediate::<_, RegistryError, _>(|transaction| {
                transaction.refresh_dispatch_singleton_record(
                    &nonce,
                    &record_json,
                    &digest,
                    updated_at,
                )
            })
            .map_err(|error| DispatchStoreError::Reconciliation(error.to_string()))?;
        Ok(())
    }

    /// Persist the primary agent's versioned SessionStart identity before any
    /// guarded tool use. Callers cannot supply binding timestamps or project
    /// identity; those facts are derived at this native boundary.
    pub fn bind_root(
        &self,
        request: BindRootDispatchRequest,
        now: i64,
    ) -> DispatchServiceResult<RootSessionBinding> {
        validate_schema(&request.schema)?;
        let run = selected_binding_run(&self.store, request.run.as_deref())?;
        let state = self.store.load_run(&run)?;
        validate_root_binding_mode(&request.mode, &state.status)?;
        let lease_expires_at = lease_expires_at(request.lease_ms, now)?;
        let binding = RootSessionBinding {
            schema: ROOT_SESSION_SCHEMA.into(),
            project_id: self.project_id.clone(),
            run,
            harness: request.harness,
            session_id: SessionId::new(request.session_id)?,
            role: Role::from_carrier(&request.role_carrier)?,
            mode: request.mode,
            project_filesystem_id: root_binding_filesystem_id(self.workspace_root.as_deref())?,
            bound_at: now,
            expires_at: lease_expires_at,
        };
        binding.validate()?;
        // An interrupted renewal is not repaired by a read or an unrelated
        // bootstrap. The exact same principal must explicitly request it and
        // re-establish current installed-carrier custody first.
        if let Some(intent) = self.store.root_renewal_intent(&binding.session_id)? {
            let normalized = self.compatible_root_request(&intent.expected, &binding)?;
            if !same_root_identity(&intent.replacement, &normalized)
                || now < intent.replacement.bound_at
            {
                return Err(DispatchServiceError::RootBindingConflict {
                    run: intent.expected.run,
                    session_id: intent.expected.session_id,
                });
            }
            self.validate_root_rebootstrap_install(
                &intent.expected,
                intent.profile_lease.as_ref(),
            )?;
            self.store.reconcile_root_renewal(&intent)?;
        }
        let published = match self.store.publish_root_binding_for_run(&binding) {
            Ok(()) => Ok(binding),
            Err(DispatchStoreError::AlreadyExists { .. }) => {
                let existing = self
                    .store
                    .load_root_binding_for_run(&binding.run, &binding.session_id)?;
                self.validate_current_root_binding(&existing)?;
                let binding = self.compatible_root_request(&existing, &binding)?;
                if same_root_identity(&existing, &binding) {
                    if now < existing.bound_at {
                        return Err(DispatchServiceError::InvalidRequest(
                            "root rebootstrap predates its current binding".into(),
                        ));
                    }
                    if binding.expires_at > existing.expires_at
                        && (self.installed_package.is_some() || now >= existing.expires_at)
                    {
                        let profile = self.root_profile_snapshot(&existing)?;
                        self.validate_root_rebootstrap_install(&existing, profile.as_ref())?;
                        self.store
                            .renew_root_binding(&existing, &binding, profile.as_ref())
                            .map_err(Into::into)
                    } else {
                        Ok(existing)
                    }
                } else if can_transition_root_to_execution(&existing, &binding) {
                    self.store
                        .transition_root_binding_to_execution(&existing, &binding)?;
                    Ok(binding)
                } else {
                    Err(DispatchServiceError::RootBindingConflict {
                        run: existing.run,
                        session_id: existing.session_id,
                    })
                }
            }
            Err(error) => Err(error.into()),
        }?;
        self.store.activate_current_root_binding(&published)?;
        Ok(published)
    }

    fn compatible_root_request(
        &self,
        existing: &RootSessionBinding,
        requested: &RootSessionBinding,
    ) -> DispatchServiceResult<RootSessionBinding> {
        let mut normalized = requested.clone();
        if existing.project_filesystem_id == requested.project_filesystem_id {
            return Ok(normalized);
        }
        normalized.project_filesystem_id = existing.project_filesystem_id.clone();
        let legacy = existing
            .project_filesystem_id
            .as_ref()
            .is_none_or(|value| value.starts_with("unix:"));
        if !legacy || !same_root_principal(existing, &normalized) {
            return Err(DispatchServiceError::RootBindingConflict {
                run: existing.run.clone(),
                session_id: existing.session_id.clone(),
            });
        }
        self.validate_root_rebootstrap_install(existing, None)?;
        if existing.project_filesystem_id.is_some() {
            validate_root_binding_filesystem(existing, self.workspace_root.as_deref())?;
        }
        Ok(normalized)
    }

    fn root_profile_snapshot(
        &self,
        root: &RootSessionBinding,
    ) -> DispatchServiceResult<Option<ProfileLease>> {
        match self.store.load_profile_lease(&root.run, &root.session_id) {
            Ok(lease) => Ok(Some(lease)),
            Err(DispatchStoreError::Io { source, .. })
                if source.kind() == std::io::ErrorKind::NotFound =>
            {
                Ok(None)
            }
            Err(error) => Err(error.into()),
        }
    }

    fn validate_root_rebootstrap_install(
        &self,
        root: &RootSessionBinding,
        profile: Option<&ProfileLease>,
    ) -> DispatchServiceResult<()> {
        let installed = self.installed_package()?;
        trusted_skill_carrier(installed, root.harness, root.role, false)?;
        if let Some(profile) = profile
            && matches!(
                profile.state,
                ProfileLeaseState::Entered | ProfileLeaseState::Active
            )
        {
            let trusted = trusted_skill_carrier(installed, root.harness, Role::Planter, false)?;
            if profile.expected_attachment.candidate_sha256 != trusted.candidate_sha256
                || profile.expected_attachment.carrier_attachment_sha256
                    != profile_carrier_digest(&trusted)
                || profile.expected_attachment.startup_skill != trusted.startup_skill
            {
                return Err(attachment_mismatch(
                    "root renewal cannot change its retained active profile carrier",
                ));
            }
        }
        Ok(())
    }

    /// Validate the explicit root bootstrap named by the host command. The
    /// hook transport owns the trusted session identity and persists the
    /// binding before this command executes; this boundary confirms only the
    /// selected run and lifecycle mode and never mints session authority from
    /// shell arguments.
    pub fn confirm_root_bootstrap(&self, run: &str, mode: &str) -> DispatchServiceResult<RunId> {
        let run = RunId::new(run)?;
        let state = self.store.load_run(&run)?;
        validate_root_binding_mode(mode, &state.status)?;
        Ok(run)
    }

    pub fn selected_run_status(&self, run: &str) -> DispatchServiceResult<Option<String>> {
        let run = RunId::new(run)?;
        Ok(self
            .store
            .load_run_if_present(&run)?
            .map(|state| state.status))
    }

    pub fn profile_enter(
        &self,
        request: ProfileEnterRequest,
        now: i64,
    ) -> DispatchServiceResult<ProfileLease> {
        validate_profile_schema(&request.schema)?;
        let run = RunId::new(request.run)?;
        let session_id = SessionId::new(request.session_id)?;
        let root = self.store.load_root_binding_for_run(&run, &session_id)?;
        self.validate_current_root_binding(&root)?;
        validate_profile_root(&root, request.harness, self.workspace_root.as_deref())?;
        let state = self.store.load_run(&run)?;
        let verified_seed_persisted = !state.seed.is_empty()
            && crate::seed_verifier::verify_persisted_seed(
                self.workspace_root.as_deref().ok_or_else(|| {
                    DispatchServiceError::InvalidRequest(
                        "profile enter requires the bound workspace root".into(),
                    )
                })?,
                &run,
                &state.seed,
            )
            .is_ok();
        let expectation = trusted_profile_attachment(self.installed_package()?, request.harness)?;
        let expires_at = lease_expires_at(request.lease_ms, now)?;
        let lease = ProfileLease::enter(
            &root,
            expectation,
            now,
            expires_at,
            &state.status,
            verified_seed_persisted,
        )?;
        self.store.publish_profile_lease(&lease)?;
        Ok(lease)
    }

    pub fn profile_activate(
        &self,
        request: ProfileActivateRequest,
        now: i64,
    ) -> DispatchServiceResult<ProfileLease> {
        validate_profile_schema(&request.schema)?;
        let run = RunId::new(request.run)?;
        let session_id = SessionId::new(request.session_id)?;
        let root = self.store.load_root_binding_for_run(&run, &session_id)?;
        self.validate_current_root_binding(&root)?;
        validate_profile_root(&root, request.harness, self.workspace_root.as_deref())?;
        let expected = self.store.load_profile_lease(&run, &session_id)?;
        let current_attachment =
            trusted_profile_attachment(self.installed_package()?, request.harness)?;
        if current_attachment.carrier_attachment_sha256
            != expected.expected_attachment.carrier_attachment_sha256
            || current_attachment.startup_skill != expected.expected_attachment.startup_skill
        {
            return Err(DispatchError::AttachmentMismatch(
                "profile carrier or planting bundle changed after enter".into(),
            )
            .into());
        }
        let mut replacement = expected.clone();
        let attestation = expected.expected_attachment.attestation();
        replacement.activate(&root, &attestation, now)?;
        self.store.replace_profile_lease(&expected, &replacement)?;
        Ok(replacement)
    }

    pub fn profile_exit(
        &self,
        request: ProfileExitRequest,
        now: i64,
    ) -> DispatchServiceResult<ProfileLease> {
        validate_profile_schema(&request.schema)?;
        let run = RunId::new(request.run)?;
        let session_id = SessionId::new(request.session_id)?;
        let root = self.store.load_root_binding_for_run(&run, &session_id)?;
        self.validate_current_root_binding(&root)?;
        validate_profile_root(&root, request.harness, self.workspace_root.as_deref())?;
        let state = self.store.load_run(&run)?;
        let workspace_root = self.workspace_root.as_deref().ok_or_else(|| {
            DispatchServiceError::InvalidRequest(
                "profile exit requires the bound workspace root".into(),
            )
        })?;
        crate::seed_verifier::verify_persisted_seed(workspace_root, &run, &state.seed).map_err(
            |error| {
                DispatchServiceError::InvalidRequest(
                    error
                        .message_text()
                        .unwrap_or("persisted seed verification failed")
                        .to_owned(),
                )
            },
        )?;
        let expected = self.store.load_profile_lease(&run, &session_id)?;
        let mut replacement = expected.clone();
        replacement.exit(&root, now, true)?;
        self.store.replace_profile_lease(&expected, &replacement)?;
        Ok(replacement)
    }

    pub fn profile_revoke(
        &self,
        request: ProfileExitRequest,
        now: i64,
    ) -> DispatchServiceResult<ProfileLease> {
        validate_profile_schema(&request.schema)?;
        let run = RunId::new(request.run)?;
        let session_id = SessionId::new(request.session_id)?;
        let root = self.store.load_root_binding_for_run(&run, &session_id)?;
        self.validate_current_root_binding(&root)?;
        validate_profile_root(&root, request.harness, self.workspace_root.as_deref())?;
        let expected = self.store.load_profile_lease(&run, &session_id)?;
        let mut replacement = expected.clone();
        replacement.revoke(&root, now)?;
        self.store.replace_profile_lease(&expected, &replacement)?;
        Ok(replacement)
    }

    pub fn skill_use_prepare(
        &self,
        request: SkillUsePrepareRequest,
        now: i64,
    ) -> DispatchServiceResult<SkillUseChallenge> {
        validate_skill_use_schema(&request.schema)?;
        let run = RunId::new(request.run)?;
        let dispatch_id = request.dispatch_id.map(AgentId::new).transpose()?;
        let session_id = SessionId::new(request.session_id)?;
        let authority = self.skill_use_authority(
            &run,
            dispatch_id.as_ref(),
            &session_id,
            request.harness,
            now,
        )?;
        let installed = self.installed_package()?;
        let trusted = self.skill_use_carrier(&authority, request.harness)?;
        if !shepherd::dispatch::allowed_skill_use(
            authority.role(),
            &trusted.startup_skill,
            &request.skill,
            request.stage,
        ) {
            return Err(DispatchError::InvalidSkillUse(
                "requested skill or stage is not allowed for this Native role".into(),
            )
            .into());
        }
        let skill_path = unique_skill_path(
            &installed.root,
            Path::new(&trusted.installed_carrier_path),
            &request.skill,
            request.harness,
        )?;
        let skill_bundle_sha256 = hash_skill_bundle(&skill_path, &request.skill)?;
        let expires_at = lease_expires_at(request.lease_ms, now)?;
        if expires_at > authority.expires_at() {
            return Err(DispatchError::InvalidSkillUse(
                "skill-use lease exceeds its Native authority lease".into(),
            )
            .into());
        }
        let challenge = SkillUseChallenge::prepare(SkillUsePrepare {
            project_id: self.project_id.clone(),
            run,
            dispatch_id,
            root_authority: authority.root().cloned(),
            session_id,
            target: request.harness,
            role: authority.role(),
            startup_skill: trusted.startup_skill,
            skill: request.skill,
            stage: request.stage,
            installed_carrier_path: trusted.installed_carrier_path,
            candidate_sha256: trusted.candidate_sha256,
            carrier_sha256: trusted.carrier_sha256,
            compiler_tree_sha256: trusted.compiler_tree_sha256,
            skill_bundle_sha256,
            nonce_sha256: crate::dispatch_broker::random_digest_32()
                .map_err(|error| DispatchServiceError::InvalidRequest(error.to_string()))?,
            prepared_at: now,
            expires_at,
        })?;
        self.store.publish_skill_use(&challenge)?;
        Ok(challenge)
    }

    pub fn skill_use_attest(
        &self,
        request: SkillUseAttestRequest,
        now: i64,
    ) -> DispatchServiceResult<SkillUseChallenge> {
        let expected = self.load_current_skill_use(request, now)?;
        let mut replacement = expected.clone();
        let attestation = LoadedSkillAttestation::from_challenge(&expected, now);
        let result = replacement.attest(&attestation, now);
        if replacement != expected {
            self.store.replace_skill_use(&expected, &replacement)?;
        }
        result?;
        Ok(replacement)
    }

    /// Recheck a completed attestation against current native authority and
    /// current installed bytes. This grants no dispatch or filesystem write.
    pub fn skill_use_verify(
        &self,
        request: SkillUseAttestRequest,
        now: i64,
    ) -> DispatchServiceResult<SkillUseChallenge> {
        let challenge = self.load_current_skill_use(request, now)?;
        if challenge.state != SkillUseState::Attested || now < challenge.prepared_at {
            return Err(DispatchError::InvalidSkillUse(
                "skill-use verification requires a completed Native attestation".into(),
            )
            .into());
        }
        if now >= challenge.expires_at {
            return Err(DispatchError::SkillUseExpired {
                expires_at: challenge.expires_at,
            }
            .into());
        }
        Ok(challenge)
    }

    fn load_current_skill_use(
        &self,
        request: SkillUseAttestRequest,
        now: i64,
    ) -> DispatchServiceResult<SkillUseChallenge> {
        validate_skill_use_schema(&request.schema)?;
        let run = RunId::new(request.run)?;
        let dispatch_id = request.dispatch_id.map(AgentId::new).transpose()?;
        let session_id = SessionId::new(request.session_id)?;
        let authority = self.skill_use_authority(
            &run,
            dispatch_id.as_ref(),
            &session_id,
            request.harness,
            now,
        )?;
        let expected = match &authority {
            SkillUseAuthority::Child(record) => {
                self.store
                    .load_skill_use(&run, &record.agent_id, &request.skill)?
            }
            SkillUseAuthority::Root(root) => {
                self.store.load_root_skill_use(&run, root, &request.skill)?
            }
        };
        if expected.target != request.harness
            || expected.session_id != session_id
            || expected.role != authority.role()
            || expected.root_authority.as_ref() != authority.root()
        {
            return Err(DispatchServiceError::InvalidRequest(
                "skill-use request does not match its current Native authority".into(),
            ));
        }
        let installed = self.installed_package()?;
        let trusted = self.skill_use_carrier(&authority, request.harness)?;
        let skill_path = unique_skill_path(
            &installed.root,
            Path::new(&trusted.installed_carrier_path),
            &expected.skill,
            expected.target,
        )?;
        if trusted.candidate_sha256 != expected.candidate_sha256
            || trusted.installed_carrier_path != expected.installed_carrier_path
            || trusted.carrier_sha256 != expected.carrier_sha256
            || trusted.compiler_tree_sha256 != expected.compiler_tree_sha256
            || trusted.startup_skill != expected.startup_skill
            || hash_skill_bundle(&skill_path, &expected.skill)? != expected.skill_bundle_sha256
        {
            return Err(DispatchError::AttachmentMismatch(
                "installed carrier or requested skill changed after prepare".into(),
            )
            .into());
        }
        // Reobserve identity after the carrier reads. Root receipts are also
        // checked under the run lock by the store on load and transition.
        let current = self.skill_use_authority(
            &run,
            dispatch_id.as_ref(),
            &session_id,
            request.harness,
            now,
        )?;
        if current != authority {
            return Err(DispatchError::InvalidSkillUse(
                "Native skill authority changed during validation".into(),
            )
            .into());
        }
        Ok(expected)
    }

    fn skill_use_authority(
        &self,
        run: &RunId,
        dispatch_id: Option<&AgentId>,
        session_id: &SessionId,
        harness: Harness,
        now: i64,
    ) -> DispatchServiceResult<SkillUseAuthority> {
        if let Some(dispatch_id) = dispatch_id {
            let record = self.store.load_for_run(run, dispatch_id)?;
            self.validate_dispatch_root_binding(&record, now)?;
            validate_skill_use_record(&record, harness, session_id, now)?;
            return Ok(SkillUseAuthority::Child(Box::new(record)));
        }
        let binding = self.store.load_root_binding_for_run(run, session_id)?;
        self.validate_current_root_binding(&binding)?;
        if binding.project_id != self.project_id || binding.harness != harness {
            return Err(DispatchError::InvalidSkillUse(
                "root skill use names a foreign Native principal".into(),
            )
            .into());
        }
        // A legacy binding without a filesystem ID remains readable. Its
        // project/run/session and installed descriptor still bind this exact
        // workspace. A present filesystem ID must always match.
        if binding.project_filesystem_id.is_some() {
            validate_root_binding_filesystem(&binding, self.workspace_root.as_deref())?;
        }
        let profile_lease = match self.store.load_profile_lease(run, session_id) {
            Ok(lease) => Some(lease),
            Err(DispatchStoreError::Io { source, .. })
                if source.kind() == std::io::ErrorKind::NotFound =>
            {
                None
            }
            Err(error) => return Err(error.into()),
        };
        let authority = SkillUseRootAuthority {
            binding,
            profile_lease,
        };
        authority.validate_live(now)?;
        let state = self.store.load_run(run)?;
        if matches!(state.status.as_str(), "closed" | "closing") {
            return Err(DispatchError::InvalidSkillUse(
                "closed run cannot issue or use skill authority".into(),
            )
            .into());
        }
        Ok(SkillUseAuthority::Root(Box::new(authority)))
    }

    fn skill_use_carrier(
        &self,
        authority: &SkillUseAuthority,
        harness: Harness,
    ) -> DispatchServiceResult<TrustedSkillCarrier> {
        let trusted = trusted_skill_carrier(
            self.installed_package()?,
            harness,
            authority.role(),
            authority.root().is_none(),
        )?;
        match authority {
            SkillUseAuthority::Child(record) => {
                validate_record_startup_attachment(record, &trusted)?
            }
            SkillUseAuthority::Root(root) => {
                if let Some(lease) = &root.profile_lease
                    && lease.state == ProfileLeaseState::Active
                    && (lease.expected_attachment.candidate_sha256 != trusted.candidate_sha256
                        || lease.expected_attachment.carrier_attachment_sha256
                            != profile_carrier_digest(&trusted)
                        || lease.expected_attachment.startup_skill != trusted.startup_skill)
                {
                    return Err(attachment_mismatch(
                        "active profile carrier differs from its Native lease",
                    ));
                }
            }
        }
        Ok(trusted)
    }

    pub fn review_ruling(
        &self,
        request: ReviewRulingRequest,
        now: i64,
    ) -> DispatchServiceResult<ReviewCustody> {
        if request.schema != REVIEW_RULING_REQUEST_SCHEMA || request.task_generation == 0 {
            return Err(DispatchServiceError::InvalidRequest(
                "review ruling request schema or task generation is invalid".into(),
            ));
        }
        request.review.validate()?;
        let run = RunId::new(request.run)?;
        let root_session_id = SessionId::new(request.root_session_id)?;
        let subject_id = AgentId::new(request.subject_agent_id)?;
        let reviewer_id = AgentId::new(&request.reviewer_dispatch_id)?;
        let reviewer_dispatch_id = DispatchId::new(request.reviewer_dispatch_id)?;
        let reviewer_session_id = SessionId::new(request.reviewer_session_id)?;
        if subject_id == reviewer_id || request.review.run != run {
            return Err(DispatchServiceError::InvalidRequest(
                "review ruling must bind distinct same-run subject and reviewer identities".into(),
            ));
        }

        let subject = self.store.load_for_run(&run, &subject_id)?;
        let reviewer = self.store.load_for_run(&run, &reviewer_id)?;
        self.validate_dispatch_root_binding(&subject, now)?;
        self.validate_dispatch_root_binding(&reviewer, now)?;
        if subject.state != DispatchState::Active
            || reviewer.state != DispatchState::Active
            || subject.root_session_id != root_session_id
            || reviewer.root_session_id != root_session_id
            || reviewer.harness != request.harness
            || reviewer.session_id != reviewer_session_id
            || reviewer.role != request.review.reviewer_role
            || !matches!(reviewer.role, Role::Critic | Role::Auditor)
            || subject.lane != request.review.lane
            || reviewer.lane != request.review.lane
        {
            return Err(DispatchServiceError::InvalidRequest(
                "review ruling does not match the live subject, reviewer, lane, carrier, or root"
                    .into(),
            ));
        }
        let pending = self.store.load_pending_for_agent(&run, &subject_id)?;
        if pending.root_session_id != root_session_id
            || pending.project_id != subject.project_id
            || pending.run != subject.run
            || pending.role != subject.role
            || pending.lane != subject.lane
            || pending.expected_attachment.agent_id != subject_id
            || pending.expected_attachment.target != subject.harness
            || pending.expected_child_session_id != subject.session_id
            || pending.launch_state != PendingLaunchState::Active
        {
            return Err(DispatchServiceError::InvalidRequest(
                "review subject has no exact active Native pending claim".into(),
            ));
        }
        if request.review.lane.is_none() {
            self.validate_planning_review_lineage(&subject, &reviewer, &pending)?;
        } else if subject.role == Role::Engineer {
            return Err(DispatchServiceError::InvalidRequest(
                "Engineer planning review is run-scoped and cannot invent a lane".into(),
            ));
        }
        let findings = serde_json::to_vec(&request.review.findings).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot encode canonical review findings: {error}"
            ))
        })?;
        let ruling = ReviewRuling {
            schema: shepherd::dispatch::REVIEW_RULING_SCHEMA.into(),
            project_id: self.project_id.clone(),
            run: run.clone(),
            subject_agent_id: subject_id.clone(),
            task_sha256: pending.task_sha256,
            task_generation: request.task_generation,
            reviewer_dispatch_id,
            findings_sha256: sha256(&findings),
            verdict: request.review.verdict,
            ruled_at: now,
        };
        ruling.validate()?;

        let (expected, mut custody) = match self.store.load_review_custody(&run, &subject_id) {
            Ok(existing) => (Some(existing.clone()), existing),
            Err(DispatchStoreError::Io { source, .. })
                if source.kind() == std::io::ErrorKind::NotFound =>
            {
                (
                    None,
                    ReviewCustody::begin(
                        &ruling,
                        root_session_id,
                        subject.session_id.clone(),
                        subject.role,
                        subject.lane.clone(),
                        pending.launch_id_hash,
                    )?,
                )
            }
            Err(error) => return Err(error.into()),
        };
        custody.apply(&ruling)?;
        if custody.state == ReviewCustodyState::Malignant {
            let (terminal, _) =
                self.store
                    .quarantine_malignant(expected.as_ref(), &subject, &pending, &custody)?;
            self.refresh_singleton_publication(&terminal, now)?;
        } else if let Some(expected) = expected.as_ref() {
            self.store.replace_review_custody(expected, &custody)?;
        } else {
            self.store.publish_review_custody(&custody)?;
        }
        Ok(custody)
    }

    fn validate_planning_review_lineage(
        &self,
        subject: &DispatchRecord,
        reviewer: &DispatchRecord,
        pending: &PendingDispatch,
    ) -> DispatchServiceResult<()> {
        let root = self
            .store
            .load_root_binding_for_run(&subject.run, &subject.root_session_id)?;
        self.validate_current_root_binding(&root)?;
        let state = self.store.load_run(&subject.run)?;
        if root.role != Role::Shepherd
            || root.mode != "planning"
            || state.status != "planted"
            || subject.role != Role::Engineer
            || subject.lane.is_some()
            || subject.parent_agent_id.is_some()
            || pending.work_kind != WorkKind::Planning
            || pending.run_status != "planted"
            || pending.caller_role != Role::Shepherd
            || pending.parent_dispatch_id.is_some()
            || pending
                .write_scope
                .iter()
                .map(PathAuthority::as_str)
                .collect::<Vec<_>>()
                != subject
                    .write_scope
                    .iter()
                    .map(String::as_str)
                    .collect::<Vec<_>>()
            || reviewer.role != Role::Critic
            || reviewer.lane.is_some()
            || !reviewer.write_scope.is_empty()
            || reviewer.parent_agent_id.as_ref() != Some(&subject.agent_id)
        {
            return Err(DispatchServiceError::InvalidRequest(
                "run-scoped review requires the planted planning root, lane-free Engineer, and its child Critic".into(),
            ));
        }
        let reviewer_pending = self
            .store
            .load_pending_for_agent(&reviewer.run, &reviewer.agent_id)?;
        if reviewer_pending.project_id != subject.project_id
            || reviewer_pending.run != subject.run
            || reviewer_pending.root_session_id != subject.root_session_id
            || reviewer_pending.run_status != "planted"
            || reviewer_pending.role != Role::Critic
            || reviewer_pending.work_kind != WorkKind::Review
            || reviewer_pending.caller_role != Role::Engineer
            || reviewer_pending
                .parent_dispatch_id
                .as_ref()
                .map(|id| id.as_str())
                != Some(subject.agent_id.as_str())
            || reviewer_pending.lane.is_some()
            || !reviewer_pending.write_scope.is_empty()
            || reviewer_pending.expected_attachment.agent_id != reviewer.agent_id
            || reviewer_pending.expected_attachment.target != reviewer.harness
            || reviewer_pending.expected_child_session_id != reviewer.session_id
            || reviewer_pending.launch_state != PendingLaunchState::Active
        {
            return Err(DispatchServiceError::InvalidRequest(
                "planning Critic has no exact active Native Engineer-owned pending claim".into(),
            ));
        }
        Ok(())
    }

    /// Observe persisted terminal custody and exercise pure prelaunch predicates.
    /// This is not a provider launch, broker-peer request, repair, or new grant.
    pub(crate) fn review_verify_terminal(
        &self,
        request: ReviewTerminalVerificationRequest,
        now: i64,
    ) -> DispatchServiceResult<serde_json::Value> {
        if request.schema != "shepherd.review-terminal-verification-request/1" {
            return Err(DispatchServiceError::InvalidRequest(
                "unsupported terminal review verification schema".into(),
            ));
        }
        let run = RunId::new(&request.run)?;
        let subject_id = AgentId::new(&request.subject_agent_id)?;
        let snapshot = self
            .store
            .read_review_terminal_snapshot(&run, &subject_id)?;
        let root = &snapshot.root;
        let subject = &snapshot.subject;
        let pending = &snapshot.pending;
        let custody = &snapshot.custody;
        self.validate_current_root_binding(root)?;
        if root.project_id != self.project_id
            || root.run != run
            || root.harness != request.harness
            || root.role != Role::Shepherd
            || root.session_id.as_str() != request.root_session_id
            || now < root.bound_at
            || now >= root.expires_at
            || subject.project_id != root.project_id
            || subject.harness != root.harness
            || subject.root_session_id != root.session_id
            || subject.session_id.as_str() != request.subject_session_id
            || subject.state != DispatchState::Malignant
            || now < subject.stopped_at.unwrap_or(i64::MAX)
            || now >= subject.lease_expires_at
            || pending.project_id != subject.project_id
            || pending.run != subject.run
            || pending.root_session_id != root.session_id
            || pending.role != subject.role
            || pending.lane != subject.lane
            || pending.expected_attachment.agent_id != subject.agent_id
            || pending.expected_attachment.target != subject.harness
            || pending.expected_child_session_id != subject.session_id
            || pending
                .write_scope
                .iter()
                .map(PathAuthority::as_str)
                .collect::<Vec<_>>()
                != subject
                    .write_scope
                    .iter()
                    .map(String::as_str)
                    .collect::<Vec<_>>()
            || pending.launch_state != PendingLaunchState::Quarantined
            || now >= pending.expires_at
            || custody.project_id != root.project_id
            || custody.run != run
            || custody.root_session_id != root.session_id
            || custody.subject_agent_id != subject.agent_id
            || custody.subject_session_id != subject.session_id
            || custody.subject_role != subject.role
            || custody.lane != subject.lane
            || custody.task_generation != request.task_generation
            || hex(&custody.task_sha256) != request.task_sha256
            || custody.task_sha256 != pending.task_sha256
            || hex(&custody.pending_launch_id_hash) != request.pending_launch_id_hash
            || custody.pending_launch_id_hash != pending.launch_id_hash
            || custody.state != ReviewCustodyState::Malignant
            || custody.rejected_revisions != 4
            || custody.rulings.len() != 4
            || !custody.claim_revoked
            || !custody.write_revoked
            || !custody.session_quarantined
            || custody.stopped_at != subject.stopped_at
            || custody.destroyed_agent_id.as_ref() != Some(&subject.agent_id)
            || custody.replacement_agent_id.is_some()
        {
            return Err(DispatchServiceError::InvalidRequest(
                "terminal review verification requires exact current-root, subject, task, and quarantined custody".into(),
            ));
        }
        // Only in-memory values are passed to the existing domain transitions.
        // Neither these arguments nor their results can be published by this path.
        let resume = subject.resume(review_resume_input(subject, now)?);
        let reclaim = validate_review_claim_custody(custody);
        let pending_claim = pending.clone().claim(
            now,
            pending.child_process_hash.ok_or_else(|| {
                DispatchServiceError::InvalidRequest(
                    "terminal pending claim has no child identity".into(),
                )
            })?,
        );
        let replay = custody.clone().apply(custody.rulings.last().ok_or_else(|| {
            DispatchServiceError::InvalidRequest("terminal review has no ruling".into())
        })?);
        if !matches!(resume, Err(DispatchError::ReviewCustodyTerminal))
            || !matches!(reclaim, Err(DispatchError::ReviewCustodyTerminal))
            || !matches!(pending_claim, Err(DispatchError::PendingLaunchConsumed))
            || !matches!(replay, Err(DispatchError::ReviewCustodyTerminal))
        {
            return Err(DispatchServiceError::InvalidRequest(
                "terminal native prelaunch predicate unexpectedly allowed a transition".into(),
            ));
        }
        self.validate_current_root_binding(root)?;
        if self
            .store
            .read_review_terminal_snapshot(&run, &subject_id)?
            != snapshot
        {
            return Err(DispatchServiceError::InvalidRequest(
                "terminal review custody changed during read-only verification".into(),
            ));
        }
        Ok(serde_json::json!({
            "schema": "shepherd.review-terminal-verification/1",
            "proof_kind": "native-prelaunch-predicate-denials",
            "broker_peer_attempted": false, "provider_launch_attempted": false, "subject_mutated": false,
            "project_id": root.project_id, "run": run, "harness": subject.harness,
            "root_session_id": root.session_id, "subject_agent_id": subject.agent_id,
            "subject_session_id": subject.session_id, "task_generation": custody.task_generation,
            "task_sha256": hex(&custody.task_sha256),
            "pending_launch_id_hash": hex(&pending.launch_id_hash),
            "dispatch_sha256": review_document_sha256(subject)?,
            "pending_sha256": review_document_sha256(pending)?,
            "custody_sha256": review_document_sha256(custody)?,
            "verified_at": now,
            "pending_claim_denial": DispatchError::PendingLaunchConsumed.to_string(),
            "denials": {
                "resume": DispatchError::ReviewCustodyTerminal.to_string(),
                "reclaim": DispatchError::ReviewCustodyTerminal.to_string(),
                "replay": DispatchError::ReviewCustodyTerminal.to_string(),
            },
        }))
    }

    pub fn review_replace(
        &self,
        request: ReviewReplacementRequest,
        now: i64,
    ) -> DispatchServiceResult<ReviewCustody> {
        if request.schema != REVIEW_REPLACEMENT_REQUEST_SCHEMA {
            return Err(DispatchServiceError::InvalidRequest(
                "review replacement request schema is invalid".into(),
            ));
        }
        let run = RunId::new(request.run)?;
        let root_session_id = SessionId::new(request.root_session_id)?;
        let subject_id = AgentId::new(request.subject_agent_id)?;
        let replacement_id = AgentId::new(request.replacement_agent_id)?;
        let root = self
            .store
            .load_root_binding_for_run(&run, &root_session_id)?;
        self.validate_current_root_binding(&root)?;
        if root.harness != request.harness
            || root.role != Role::Shepherd
            || now < root.bound_at
            || now >= root.expires_at
        {
            return Err(DispatchServiceError::InvalidRequest(
                "review replacement requires the exact live Shepherd root".into(),
            ));
        }
        let expected = self.store.load_review_custody(&run, &subject_id)?;
        if expected.state != ReviewCustodyState::Malignant
            || expected.root_session_id != root_session_id
        {
            return Err(DispatchServiceError::Domain(
                DispatchError::ReviewCustodyTerminal,
            ));
        }
        let original = self.store.load_pending_for_agent(&run, &subject_id)?;
        let state = self.store.load_run(&run)?;
        let mode_permits_replacement = match (root.mode.as_str(), state.status.as_str()) {
            ("planning", "planted") => {
                original.run_status == "planted"
                    && original.role == Role::Engineer
                    && original.work_kind == WorkKind::Planning
                    && original.lane.is_none()
                    && original.caller_role == Role::Shepherd
                    && original.parent_dispatch_id.is_none()
            }
            ("execution", "executing") => original.run_status == "executing",
            _ => false,
        };
        if !mode_permits_replacement
            || original.project_id != root.project_id
            || expected.subject_role != original.role
            || expected.lane != original.lane
            || expected.subject_session_id != original.expected_child_session_id
            || expected.subject_agent_id != original.expected_attachment.agent_id
            || expected.task_sha256 != original.task_sha256
            || expected.pending_launch_id_hash != original.launch_id_hash
        {
            return Err(DispatchServiceError::InvalidRequest(
                "review replacement does not match the current root phase and malignant subject contract".into(),
            ));
        }
        let replacement = self.store.load_pending_for_agent(&run, &replacement_id)?;
        if replacement.launch_state != PendingLaunchState::Pending
            || replacement.root_session_id != root_session_id
            || replacement.caller_role != Role::Shepherd
            || replacement.parent_dispatch_id.is_some()
            || replacement.replaces_agent_id.as_ref() != Some(&subject_id)
            || replacement.expected_child_session_id == original.expected_child_session_id
            || !same_replacement_contract(&original, &replacement)
        {
            return Err(DispatchServiceError::InvalidRequest(
                "replacement pending claim changed the malignant subject task or scope contract"
                    .into(),
            ));
        }
        let mut custody = expected.clone();
        custody.replace(replacement_id, now)?;
        self.store.replace_review_custody(&expected, &custody)?;
        Ok(custody)
    }

    pub fn resolve(
        &self,
        request: ResolveDispatchRequest,
        now: i64,
    ) -> DispatchServiceResult<DispatchResolution> {
        validate_schema(&request.schema)?;
        let role = request
            .role_carrier
            .as_deref()
            .map(Role::from_carrier)
            .transpose()?;
        let agent_id = request.agent_id.map(AgentId::new).transpose()?;
        let agent_type = request.agent_type.map(AgentType::new).transpose()?;
        let lane = request.lane.map(LaneId::new).transpose()?;
        let session_id = SessionId::new(request.session_id)?;
        let tool_name = request.tool_name;
        let tool_input = request.tool_input;
        let (run, root_binding, profile_lease) = if agent_id.is_none() {
            let binding = match request.run.as_deref() {
                Some(requested) => {
                    let requested = RunId::new(requested)?;
                    self.store
                        .load_root_binding_for_run(&requested, &session_id)?
                }
                None => self.store.load_current_root_binding(&session_id)?,
            };
            self.validate_current_root_binding(&binding)?;
            validate_root_binding_filesystem(&binding, self.workspace_root.as_deref())?;
            validate_requested_binding_run(request.run.as_deref(), &binding.run)?;
            let state = self.store.load_run(&binding.run)?;
            validate_root_binding_mode(&binding.mode, &state.status)?;
            let profile = self.live_profile_for_root(&binding, now)?;
            (binding.run.clone(), Some(binding), profile)
        } else {
            // The run is a selector, never authority: the native record and
            // its current root binding are validated below. Planning children
            // must not depend on global executing-run discovery or siblings.
            let selected = selected_binding_run(&self.store, request.run.as_deref())?;
            let state = self.store.load_run(&selected)?;
            if !matches!(
                state.status.as_str(),
                "planted" | "planned" | "executing" | "integrating" | "closing"
            ) {
                return Err(DispatchServiceError::InvalidRequest(format!(
                    "child tools are unavailable for run status `{}`",
                    state.status
                )));
            }
            (selected, None, None)
        };
        let tool_call_id = request.tool_call_id;
        let native = NativeIdentity {
            harness: request.harness,
            project_id: self.project_id.clone(),
            run: run.clone(),
            lane,
            session_id: session_id.clone(),
            agent_id,
            agent_type,
            role,
            tool_call_id: tool_call_id.clone(),
            now,
            root_binding,
        };
        let (resolution, record) = self.store.resolve_identity_for_run_with_record(&native)?;
        if let Some(record) = record.as_ref() {
            self.validate_dispatch_root_binding(record, now)?;
        }
        let mut resolution = match resolution {
            IdentityResolution::Root { role, mode } => DispatchResolution {
                schema: "shepherd.identity-resolution/1".into(),
                project_id: self.project_id.clone(),
                run,
                harness: request.harness,
                agent_id: None,
                agent_type: None,
                role: profile_lease.as_ref().map_or(role, |_| Role::Planter),
                lane: None,
                session_id,
                // A root binding has no dispatch lane or requested file scope.
                // Shepherd's structured write tools remain root-Markdown only;
                // exact-root Bash recovery is authorized separately because
                // shell text is never treated as path authority.
                write_scope: profile_lease.as_ref().map_or_else(
                    || vec![String::from("*.md")],
                    |lease| lease.write_scope.iter().map(ToString::to_string).collect(),
                ),
                capabilities: None,
                tool_call_id,
                mode: Some(mode),
                write_paths: Vec::new(),
                path_in_write_scope: None,
            },
            IdentityResolution::Agent { agent_id, role } => DispatchResolution {
                schema: "shepherd.identity-resolution/1".into(),
                project_id: self.project_id.clone(),
                run,
                harness: request.harness,
                agent_id: Some(agent_id.to_string()),
                agent_type: record.as_ref().map(|record| record.agent_type.clone()),
                role,
                lane: record.as_ref().and_then(|record| record.lane.clone()),
                session_id,
                write_scope: record
                    .as_ref()
                    .map(|record| record.write_scope.clone())
                    .unwrap_or_default(),
                capabilities: record.map(|record| record.capabilities),
                tool_call_id,
                mode: None,
                write_paths: Vec::new(),
                path_in_write_scope: None,
            },
        };
        if tool_name.is_some() || tool_input.is_some() {
            let workspace_root = self.workspace_root.as_deref().ok_or_else(|| {
                DispatchServiceError::InvalidRequest(
                    "native write-path resolution requires the bound workspace root".into(),
                )
            })?;
            let exact_root = resolution.agent_id.is_none()
                && resolution.lane.is_none()
                && matches!(resolution.role, Role::Shepherd | Role::Planter)
                && matches!(resolution.mode.as_deref(), Some("planning" | "execution"));
            let write_paths = derive_write_paths(
                workspace_root,
                tool_name.as_deref(),
                tool_input.as_ref(),
                exact_root,
            )?;
            if !write_paths.is_empty() {
                let mut all_in_scope = true;
                for path in &write_paths {
                    all_in_scope &= path_in_write_scope(path, &resolution.write_scope)?;
                }
                resolution.path_in_write_scope = Some(all_in_scope);
                resolution.write_paths = write_paths;
            }
        }
        Ok(resolution)
    }

    fn live_profile_for_root(
        &self,
        root: &RootSessionBinding,
        now: i64,
    ) -> DispatchServiceResult<Option<ProfileLease>> {
        let lease = match self.store.load_profile_lease(&root.run, &root.session_id) {
            Ok(lease) => lease,
            Err(DispatchStoreError::Io { source, .. })
                if source.kind() == std::io::ErrorKind::NotFound =>
            {
                return Ok(None);
            }
            Err(error) => return Err(error.into()),
        };
        if matches!(
            lease.state,
            ProfileLeaseState::Exited | ProfileLeaseState::Revoked
        ) {
            return Ok(None);
        }
        if now >= lease.expires_at || now >= root.expires_at {
            let expected = lease.clone();
            let mut revoked = lease;
            revoked.revoke(root, now)?;
            self.store.replace_profile_lease(&expected, &revoked)?;
            return Ok(None);
        }
        lease.validate_for_root(root, now)?;
        Ok((lease.state == ProfileLeaseState::Active).then_some(lease))
    }

    pub fn resolve_for_mutation(
        &self,
        agent_id: &str,
        now: i64,
    ) -> DispatchServiceResult<DispatchRecord> {
        let agent_id = AgentId::new(agent_id)?;
        let record = self.store.load_active(&agent_id)?;
        self.validate_dispatch_root_binding(&record, now)?;
        if record.state != DispatchState::Active {
            return Err(IdentityError::Terminal {
                state: record.state,
            }
            .into());
        }
        if now >= record.lease_expires_at {
            return Err(IdentityError::Stale {
                expired_at: record.lease_expires_at,
            }
            .into());
        }
        if record.capabilities.readiness() == CapabilityReadiness::Blocked {
            return Err(IdentityError::CapabilityBlocked.into());
        }
        Ok(record)
    }

    pub fn stop(
        &self,
        request: StopDispatchRequest,
        now: i64,
    ) -> DispatchServiceResult<DispatchRecord> {
        validate_schema(&request.schema)?;
        let run = selected_binding_run(&self.store, request.run.as_deref())?;
        let agent_id = AgentId::new(request.agent_id)?;
        let native = NativeIdentity {
            harness: request.harness,
            project_id: self.project_id.clone(),
            run,
            lane: request.lane.map(LaneId::new).transpose()?,
            session_id: SessionId::new(request.session_id)?,
            agent_id: Some(agent_id.clone()),
            agent_type: Some(AgentType::new(request.agent_type)?),
            role: request
                .role_carrier
                .as_deref()
                .map(Role::from_carrier)
                .transpose()?,
            tool_call_id: None,
            now,
            root_binding: None,
        };
        let current = self.store.load_for_run(&native.run, &agent_id)?;
        self.validate_dispatch_root_binding(&current, now)?;
        let record = self.store.stop_verified_for_run(
            &native,
            StopRequest {
                agent_id,
                expected_revision: request.expected_revision,
                stopped_at: now,
                result_artifact: request.result_artifact,
            },
        )?;
        self.refresh_singleton_publication(&record, now)?;
        Ok(record)
    }

    pub fn project_filesystem_id(&self) -> DispatchServiceResult<ProjectFilesystemId> {
        let root = self.workspace_root.as_deref().ok_or_else(|| {
            DispatchServiceError::InvalidRequest(
                "pending dispatch requires the bound workspace root".into(),
            )
        })?;
        filesystem_id(root)
    }

    pub(crate) fn prepare_pending(
        &self,
        parent: &crate::dispatch_broker::BrokerParent,
        request: PreparePendingDispatchRequest,
        launch_id: BrokerLaunchId,
        launch_hash: [u8; 32],
        now: i64,
    ) -> DispatchServiceResult<PreparePendingDispatchResponse> {
        validate_pending_schema(&request.schema)?;
        let lease_ms = request.lease_ms;
        let run = selected_broker_run(&self.store, &request.run, parent)?;
        let state = self.validate_broker_parent(parent, &run, now)?;
        let role = parse_role(&request.role)?;
        let work_kind = parse_work_kind(&request.work_kind)?;
        if role == Role::Engineer {
            let root = self.workspace_root.as_deref().ok_or_else(|| {
                DispatchServiceError::InvalidRequest(
                    "Engineer preparation requires the bound workspace root".into(),
                )
            })?;
            crate::seed_verifier::verify_persisted_seed_with_state(root, &run, &state.seed, &state)
                .map_err(|error| {
                    DispatchServiceError::InvalidRequest(format!(
                        "Engineer remains blocked until the persisted seed verifies: {}",
                        error
                            .message_text()
                            .unwrap_or("persisted seed verification failed")
                    ))
                })?;
        }
        validate_pending_edge(&state.status, parent.role, role, work_kind)?;
        validate_parent_dispatch(
            parent.dispatch_id.as_ref(),
            request.parent_dispatch_id.as_deref(),
        )?;
        let lane = request.lane.map(LaneId::new).transpose()?;
        let replaces_agent_id = request.replaces_agent_id.map(AgentId::new).transpose()?;
        self.ensure_singleton_is_unclaimed(
            &run,
            role,
            lane.as_ref(),
            replaces_agent_id
                .as_ref()
                .map(|agent| (agent, &parent.root_session_id)),
            now,
        )?;
        if parent.dispatch_id.is_some() {
            let parent_agent = parent
                .dispatch_id
                .as_ref()
                .map(|id| AgentId::new(id.as_str()))
                .transpose()?;
            let record = self.store.load_for_run(
                &run,
                &parent_agent.ok_or_else(|| {
                    DispatchServiceError::InvalidRequest(
                        "child parent identity is incomplete".into(),
                    )
                })?,
            )?;
            if record.lane != lane {
                return Err(DispatchServiceError::InvalidRequest(
                    "pending lane does not match the authenticated parent process".into(),
                ));
            }
        }
        let parent_dispatch_id = request
            .parent_dispatch_id
            .as_deref()
            .map(shepherd::dispatch::DispatchId::new)
            .transpose()?;
        if replaces_agent_id.is_some()
            && (parent.role != Role::Shepherd || parent.dispatch_id.is_some())
        {
            return Err(DispatchServiceError::InvalidRequest(
                "replacement lineage can only be prepared by the bound root Shepherd".into(),
            ));
        }
        let baseline_commit = GitCommit::new(request.baseline)?;
        if current_git_commit(self.workspace_root.as_deref())? != baseline_commit {
            return Err(DispatchServiceError::InvalidRequest(
                "pending baseline commit does not match the native repository".into(),
            ));
        }
        let read_scope = request
            .read_scope
            .into_iter()
            .map(PathAuthority::new)
            .collect::<Result<Vec<_>, _>>()?;
        let write_scope = request
            .write_scope
            .into_iter()
            .map(PathAuthority::new)
            .collect::<Result<Vec<_>, _>>()?;
        let result_artifact = PathAuthority::exact(request.result_artifact)?;
        let review_artifact = PathAuthority::exact(request.review_artifact)?;
        validate_role_artifact_namespaces(
            &run,
            role,
            lane.as_ref(),
            &result_artifact,
            &review_artifact,
        )?;
        validate_native_scopes(
            self.workspace_root.as_deref(),
            NativeScopePolicy {
                run: &run,
                role,
                work_kind,
                lane: lane.as_ref(),
                result_artifact: &result_artifact,
            },
            &read_scope,
            &write_scope,
        )?;
        validate_run_artifact(&run, &result_artifact)?;
        validate_run_artifact(&run, &review_artifact)?;
        validate_artifact_nofollow(self.workspace_root.as_deref(), &result_artifact)?;
        validate_artifact_nofollow(self.workspace_root.as_deref(), &review_artifact)?;
        validate_scope_nofollow(self.workspace_root.as_deref(), &result_artifact)?;
        validate_scope_nofollow(self.workspace_root.as_deref(), &review_artifact)?;
        let task_path =
            repository_relative_path(self.workspace_root.as_deref(), &request.task_file)?;
        let task_authority = PathAuthority::exact(task_path.clone())?;
        let task_bytes = read_relative_nofollow(
            self.workspace_root.as_deref(),
            &task_path,
            MAX_PENDING_BYTES,
        )?;
        if !read_scope
            .iter()
            .any(|scope| scope.contains(&task_path).unwrap_or(false))
        {
            return Err(DispatchServiceError::InvalidRequest(
                "task file is outside the pending read scope".into(),
            ));
        }
        let expected_attachment =
            trusted_attachment_expectation(self.installed_package()?, request.expected_attachment)?;
        if expected_attachment.role != role {
            return Err(DispatchServiceError::Domain(
                DispatchError::AttachmentMismatch(
                    "trusted attachment role differs from pending role".into(),
                ),
            ));
        }
        if expected_attachment.target != parent.harness {
            return Err(DispatchServiceError::Domain(
                DispatchError::AttachmentMismatch(
                    "trusted attachment target differs from the authenticated parent harness"
                        .into(),
                ),
            ));
        }
        if parent
            .dispatch_id
            .as_ref()
            .is_some_and(|parent_id| parent_id.as_str() == expected_attachment.agent_id.as_str())
        {
            return Err(DispatchServiceError::Domain(DispatchError::InvalidParent(
                "child agent identity must differ from the authenticated parent".into(),
            )));
        }
        let child_session_id = SessionId::new(request.child_session_id)?;
        if child_session_id == parent.session_id {
            return Err(DispatchServiceError::InvalidRequest(
                "child session must differ from the authenticated parent session".into(),
            ));
        }
        let nonce_sha256 = crate::dispatch_broker::random_digest_32()
            .map_err(|error| DispatchServiceError::InvalidRequest(error.to_string()))?;
        let pending = PendingDispatch {
            schema: shepherd::dispatch::PENDING_DISPATCH_SCHEMA.into(),
            launch_id_hash: launch_hash,
            parent_process_hash: parent.process.digest(),
            project_id: self.project_id.clone(),
            project_filesystem_id: filesystem_id(self.workspace_root.as_deref().ok_or_else(
                || {
                    DispatchServiceError::InvalidRequest(
                        "pending dispatch requires the bound workspace root".into(),
                    )
                },
            )?)?,
            run: run.clone(),
            run_status: state.status.clone(),
            root_session_id: parent.root_session_id.clone(),
            caller_role: parent.role,
            parent_dispatch_id,
            replaces_agent_id,
            role,
            work_kind,
            lane,
            baseline_commit,
            read_scope,
            write_scope,
            result_artifact,
            review_artifact,
            task_path: task_authority,
            task_sha256: sha256(&task_bytes),
            expected_child_session_id: child_session_id,
            expected_attachment,
            expires_at: lease_expires_at(lease_ms, now)?,
            launch_state: PendingLaunchState::Pending,
            claimed_at: None,
            child_process_hash: None,
            activated_at: None,
            nonce_sha256,
        };
        pending.validate()?;
        let pending =
            self.store
                .publish_pending_bounded(&pending, lease_ms, |pending, now, authority| {
                    self.validate_planning_critic_pre(pending, now, authority)
                        .map_err(store_domain_error)?;
                    self.authorize_dispatch_budget(pending, now, authority)
                        .map_err(store_domain_error)
                })?;
        Ok(PreparePendingDispatchResponse {
            schema: "shepherd.pending-dispatch-response/2".into(),
            launch_id,
            pending,
        })
    }

    fn ensure_singleton_is_unclaimed(
        &self,
        run: &RunId,
        role: Role,
        lane: Option<&LaneId>,
        replacement: Option<(&AgentId, &SessionId)>,
        now: i64,
    ) -> DispatchServiceResult<()> {
        if !matches!(role, Role::Engineer | Role::Conductor) {
            return Ok(());
        }
        let registry_path = self.registry_path.as_ref().ok_or_else(|| {
            DispatchServiceError::InvalidRequest(
                "lead-role preparation requires the authoritative registry".into(),
            )
        })?;
        let registry = Registry::open_migrated(registry_path)
            .map_err(|error| DispatchStoreError::Reconciliation(error.to_string()))?;
        let lane_key = if role == Role::Engineer {
            "__run__"
        } else {
            lane.ok_or_else(|| {
                DispatchServiceError::InvalidRequest("Conductor preparation requires a lane".into())
            })?
            .as_str()
        };
        let claim = registry
            .load_dispatch_singleton(
                self.project_id.as_str(),
                run.as_str(),
                role.as_str(),
                lane_key,
            )
            .map_err(|error| DispatchStoreError::Reconciliation(error.to_string()))?;
        if let (Some(claim), Some((subject, root_session))) = (&claim, replacement) {
            let source = self.store.read_review_terminal_snapshot(run, subject)?;
            self.validate_current_root_binding(&source.root)?;
            let mut bytes = serde_json::to_vec(&source.subject)
                .map_err(|error| DispatchError::InvalidRecord(error.to_string()))?;
            bytes.push(b'\n');
            if claim.agent_id == subject.as_str()
                && claim.publication_state
                    == Some(shepherd::registry::SingletonPublicationState::Published)
                && claim.record_sha256.as_deref() == Some(hex(&sha256(&bytes)).as_str())
                && source.custody.state == ReviewCustodyState::Malignant
                && source.root.session_id == *root_session
                && source.root.role == Role::Shepherd
                && source.root.project_id == self.project_id
                && source.root.run == *run
                && now >= source.root.bound_at
                && now < source.root.expires_at
                && source.subject.state == DispatchState::Malignant
                && source.subject.role == role
                && source.subject.lane.as_ref() == lane
                && source.subject.root_session_id == *root_session
                && source.pending.launch_state == PendingLaunchState::Quarantined
                && root_replacement_phase(&source.root, &source.pending, &source.pending.run_status)
            {
                return Ok(());
            }
        }
        if claim.is_some() || replacement.is_some() {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "{role} singleton is already claimed or lacks exact malignant replacement lineage for run `{run}` and lane `{lane_key}`"
            )));
        }
        Ok(())
    }

    pub(crate) fn claim_pending(
        &self,
        child: &crate::dispatch_broker::BrokerChild,
        request: ClaimPendingDispatchRequest,
    ) -> DispatchServiceResult<ClaimPendingDispatchResponse> {
        validate_pending_schema(&request.schema)?;
        if request.launch_id != child.launch_id {
            return Err(DispatchServiceError::Domain(DispatchError::InvalidPending(
                "broker launch identity does not match the child connection".into(),
            )));
        }
        let pending = self.store.claim_pending_unspawned(
            &child.run,
            child.launch_hash,
            child.process.digest(),
            |pending, transition_now, authority| {
                self.validate_child_claim(child, &request, pending, transition_now, authority)
                    .map(|_| ())
                    .map_err(store_domain_error)
            },
        )?;
        Ok(ClaimPendingDispatchResponse {
            schema: "shepherd.pending-dispatch-claim/2".into(),
            launch_id: request.launch_id,
            pending,
        })
    }

    pub(crate) fn activate_pending(
        &self,
        child: &crate::dispatch_broker::BrokerChild,
        request: ClaimPendingDispatchRequest,
    ) -> DispatchServiceResult<DispatchRecord> {
        validate_pending_schema(&request.schema)?;
        if request.launch_id != child.launch_id {
            return Err(DispatchServiceError::Domain(DispatchError::InvalidPending(
                "broker launch identity does not match the child connection".into(),
            )));
        }
        let run_state = self.store.load_run(&child.run)?;
        let mut prepared_singleton = None;
        let mut is_review_replacement = false;
        let record = self.store.activate_pending(
            &child.run,
            child.launch_hash,
            child.process.digest(),
            |pending, transition_now, authority| {
                let replacement = self
                    .validate_child_claim(child, &request, pending, transition_now, authority)
                    .map_err(store_domain_error)?;
                is_review_replacement = replacement.is_some();
                let contract = pending.role.dispatch_capability_contract()?;
                let observed: BTreeSet<String> = contract
                    .required
                    .union(&contract.optional)
                    .cloned()
                    .collect();
                let record = DispatchRecord::start(DispatchStart {
                    project_id: pending.project_id.clone(),
                    run: pending.run.clone(),
                    root_session_id: pending.root_session_id.clone(),
                    run_incarnation: run_state.run_incarnation.clone(),
                    nonce: crate::native_authority::new_nonce(),
                    harness: pending.expected_attachment.target,
                    agent_id: pending.expected_attachment.agent_id.clone(),
                    agent_type: AgentType::new(request.agent_type.clone())?,
                    role: pending.role,
                    lane: pending.lane.clone(),
                    parent_agent_id: pending
                        .parent_dispatch_id
                        .as_ref()
                        .map(|id| AgentId::new(id.as_str()))
                        .transpose()?,
                    session_id: SessionId::new(request.session_id.clone())?,
                    write_scope: pending
                        .write_scope
                        .iter()
                        .map(|scope| scope.as_str().to_owned())
                        .collect(),
                    model: None,
                    capability_contract: contract,
                    capability_probe: CapabilityProbe::new(
                        observed,
                        "native-broker-activation",
                        "native",
                        None,
                        transition_now,
                    )?,
                    startup_attachment: Some(StartupAttachment {
                        skill: pending.expected_attachment.startup_skill.clone(),
                        bundle_digest: hex(&pending.expected_attachment.skill_bundle_sha256),
                    }),
                    attachment_nonce: Some(hex(&pending.nonce_sha256)),
                    result_artifact: Some(pending.result_artifact.as_str().to_owned()),
                    result_nonce: Some(crate::native_authority::new_nonce()),
                    review_artifact: matches!(pending.role, Role::Auditor | Role::Critic)
                        .then(|| pending.review_artifact.as_str().to_owned()),
                    review_nonce: matches!(pending.role, Role::Auditor | Role::Critic)
                        .then(crate::native_authority::new_nonce),
                    started_at: transition_now,
                    lease_expires_at: pending.expires_at,
                    resumes_agent_id: None,
                })?;
                prepared_singleton = self.prepare_singleton_publication(
                    &record,
                    transition_now,
                    replacement.as_ref(),
                )?;
                Ok(record)
            },
        )?;
        if let Some((mut registry, publication)) = prepared_singleton {
            if is_review_replacement {
                self.store
                    .publish_review_replacement_prepared(&publication, &mut registry)?;
            } else {
                self.store.publish_singleton_prepared(&publication)?;
                registry
                    .transaction_immediate::<_, RegistryError, _>(|transaction| {
                        transaction.mark_dispatch_singleton_published(
                            &publication.nonce,
                            record.started_at,
                        )
                    })
                    .map_err(|error| DispatchStoreError::Reconciliation(error.to_string()))?;
            }
        }
        Ok(record)
    }

    pub(crate) fn complete_broker_child(
        &self,
        child: &crate::dispatch_broker::BrokerChild,
        agent_id: String,
        session_id: String,
        agent_type: String,
        nonce_sha256: [u8; 32],
    ) -> DispatchServiceResult<DispatchRecord> {
        if !constant_time_digest_eq(&nonce_sha256, &child.nonce_sha256) {
            return Err(DispatchServiceError::Domain(DispatchError::InvalidPending(
                "broker completion nonce does not match the child connection".into(),
            )));
        }
        let agent_id = AgentId::new(agent_id)?;
        if agent_id != child.expected_attachment.agent_id {
            return Err(DispatchServiceError::Domain(DispatchError::InvalidPending(
                "broker completion agent does not match the prepared child".into(),
            )));
        }
        let session_id = SessionId::new(session_id)?;
        let pending = self.store.load_pending(&child.run, child.launch_hash)?;
        if pending.launch_state != PendingLaunchState::Active
            || pending.expected_attachment.agent_id != agent_id
            || pending.expected_child_session_id != session_id
        {
            return Err(DispatchServiceError::Domain(DispatchError::InvalidPending(
                "broker completion does not match the active prepared child".into(),
            )));
        }
        let record = self.store.load_for_run(&child.run, &agent_id)?;
        self.validate_dispatch_root_binding(&record, crate::dispatch_broker::now_millis())?;
        if record.state != DispatchState::Active
            || record.harness != pending.expected_attachment.target
            || record.session_id != session_id
        {
            return Err(DispatchServiceError::Domain(DispatchError::InvalidPending(
                "broker completion requires the exact active child record".into(),
            )));
        }
        self.stop(
            StopDispatchRequest {
                schema: REQUEST_SCHEMA.into(),
                run: Some(child.run.to_string()),
                harness: record.harness,
                agent_id: agent_id.to_string(),
                agent_type,
                role_carrier: Some(pending.role.carrier()),
                lane: record.lane.as_ref().map(ToString::to_string),
                session_id: session_id.to_string(),
                expected_revision: record.revision,
                result_artifact: Some(pending.result_artifact.as_str().to_owned()),
            },
            crate::dispatch_broker::now_millis(),
        )
    }

    fn validate_broker_parent(
        &self,
        parent: &crate::dispatch_broker::BrokerParent,
        run: &RunId,
        now: i64,
    ) -> DispatchServiceResult<shepherd::RunState> {
        let state = self.store.load_run(run)?;
        if !matches!(
            state.status.as_str(),
            "planted" | "planned" | "executing" | "integrating" | "closing"
        ) {
            return Err(DispatchServiceError::InvalidRequest(
                "broker parent is bound to a non-dispatchable run".into(),
            ));
        }
        let root = self
            .store
            .load_root_binding_for_run(run, &parent.root_session_id)?;
        self.validate_current_root_binding(&root)?;
        if root.project_id != self.project_id
            || root.harness != parent.harness
            || now >= root.expires_at
        {
            return Err(DispatchServiceError::InvalidRequest(
                "broker parent root binding is not current".into(),
            ));
        }
        if parent.dispatch_id.is_none() {
            if root.session_id != parent.session_id || root.role != parent.role {
                return Err(DispatchServiceError::InvalidRequest(
                    "broker parent does not match the registered root session".into(),
                ));
            }
        } else {
            let parent_dispatch_id = parent.dispatch_id.as_ref().ok_or_else(|| {
                DispatchServiceError::InvalidRequest(
                    "broker parent dispatch identity is missing".into(),
                )
            })?;
            let agent = AgentId::new(parent_dispatch_id.as_str())?;
            let record = self.store.load_for_run(run, &agent)?;
            if record.state != DispatchState::Active
                || record.harness != parent.harness
                || record.session_id != parent.session_id
                || record.role != parent.role
                || now >= record.lease_expires_at
            {
                return Err(DispatchServiceError::InvalidRequest(
                    "broker parent does not match the active dispatch".into(),
                ));
            }
        }
        Ok(state)
    }

    fn validate_child_claim(
        &self,
        child: &crate::dispatch_broker::BrokerChild,
        request: &ClaimPendingDispatchRequest,
        pending: &PendingDispatch,
        now: i64,
        authority: &crate::dispatch_store::LockedDispatchRun<'_>,
    ) -> DispatchServiceResult<Option<crate::dispatch_store::ReviewTerminalSnapshot>> {
        if pending.project_id != self.project_id
            || !constant_time_digest_eq(&pending.parent_process_hash, &child.parent_process_hash)
            || !constant_time_digest_eq(&pending.launch_id_hash, &child.launch_hash)
            || pending.expected_child_session_id.as_str() != request.session_id
            || pending.expected_attachment.agent_id.as_str() != request.agent_id
        {
            return Err(DispatchServiceError::Domain(DispatchError::InvalidPending(
                "child launch facts do not match the broker-prepared constraints".into(),
            )));
        }
        let root = authority.load_root_binding(&pending.root_session_id)?;
        self.validate_current_root_binding(&root)?;
        if now < root.bound_at || now >= root.expires_at {
            return Err(DispatchServiceError::InvalidRequest(
                "pending child root binding is no longer current".into(),
            ));
        }
        match authority.load_review_custody(&pending.expected_attachment.agent_id) {
            Ok(custody) => validate_review_claim_custody(&custody)?,
            Err(DispatchStoreError::Io { source, .. })
                if source.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(error.into()),
        }
        let replacement = self.validate_review_replacement_claim(pending, &root, now, authority)?;
        self.validate_planning_critic_pre(pending, now, authority)?;
        validate_pending_edge(
            &pending.run_status,
            pending.caller_role,
            pending.role,
            pending.work_kind,
        )?;
        let current_filesystem =
            filesystem_id(self.workspace_root.as_deref().ok_or_else(|| {
                DispatchServiceError::InvalidRequest(
                    "pending dispatch requires the bound workspace root".into(),
                )
            })?)?;
        if current_filesystem != pending.project_filesystem_id {
            return Err(DispatchServiceError::InvalidRequest(
                "project filesystem identity changed before child activation".into(),
            ));
        }
        if current_git_commit(self.workspace_root.as_deref())? != pending.baseline_commit {
            return Err(DispatchServiceError::InvalidRequest(
                "pending baseline commit changed before child activation".into(),
            ));
        }
        validate_role_artifact_namespaces(
            &pending.run,
            pending.role,
            pending.lane.as_ref(),
            &pending.result_artifact,
            &pending.review_artifact,
        )?;
        validate_native_scopes(
            self.workspace_root.as_deref(),
            NativeScopePolicy {
                run: &pending.run,
                role: pending.role,
                work_kind: pending.work_kind,
                lane: pending.lane.as_ref(),
                result_artifact: &pending.result_artifact,
            },
            &pending.read_scope,
            &pending.write_scope,
        )?;
        for artifact in [
            &pending.task_path,
            &pending.result_artifact,
            &pending.review_artifact,
        ] {
            validate_scope_nofollow(self.workspace_root.as_deref(), artifact)?;
        }
        let task_bytes = read_relative_nofollow(
            self.workspace_root.as_deref(),
            pending.task_path.as_str(),
            MAX_PENDING_BYTES,
        )?;
        let task_hash = sha256(&task_bytes);
        if !constant_time_digest_eq(&task_hash, &pending.task_sha256) {
            return Err(DispatchServiceError::Domain(DispatchError::InvalidPending(
                "task bytes changed before child activation".into(),
            )));
        }
        request
            .attestation
            .validate_against(&pending.expected_attachment, &pending.nonce_sha256)?;
        validate_installed_attachment(
            &request.attestation,
            &pending.expected_attachment,
            self.installed_package()?,
        )?;
        if !agent_type_matches(
            pending.expected_attachment.target,
            pending.role,
            &AgentType::new(request.agent_type.clone())?,
        ) {
            return Err(DispatchServiceError::Domain(
                DispatchError::AttachmentMismatch(
                    "child agent type does not match the trusted role carrier".into(),
                ),
            ));
        }
        if now < 0 || now >= pending.expires_at {
            return Err(DispatchServiceError::Domain(
                DispatchError::PendingExpired {
                    expires_at: pending.expires_at,
                },
            ));
        }
        self.authorize_dispatch_budget(pending, now, authority)?;
        Ok(replacement)
    }

    fn validate_planning_critic_pre(
        &self,
        pending: &PendingDispatch,
        now: i64,
        authority: &crate::dispatch_store::LockedDispatchRun<'_>,
    ) -> DispatchServiceResult<()> {
        if pending.role != Role::Critic || pending.run_status != "planted" {
            return Ok(());
        }
        let denied = || {
            DispatchServiceError::InvalidRequest(
            "planning Critic requires its exact live Engineer and accepted native orientation pre".into(),
        )
        };
        let parent = pending.parent_dispatch_id.as_ref().ok_or_else(denied)?;
        let inventory = authority.inventory()?;
        let mut engineers = inventory
            .records
            .iter()
            .filter(|record| record.agent_id.as_str() == parent.as_str());
        let engineer = engineers.next().ok_or_else(denied)?;
        let mut launches = inventory
            .pending
            .iter()
            .filter(|launch| launch.expected_attachment.agent_id == engineer.agent_id);
        let engineer_pending = launches.next().ok_or_else(denied)?;
        let root = authority.load_root_binding(&pending.root_session_id)?;
        self.validate_current_root_binding(&root)?;
        if engineers.next().is_some()
            || launches.next().is_some()
            || pending.caller_role != Role::Engineer
            || pending.work_kind != WorkKind::Review
            || pending.lane.is_some()
            || engineer.state != DispatchState::Active
            || engineer.role != Role::Engineer
            || engineer.lane.is_some()
            || engineer.project_id != pending.project_id
            || engineer.run != pending.run
            || engineer.root_session_id != pending.root_session_id
            || engineer.harness != pending.expected_attachment.target
            || now < engineer.started_at
            || now >= engineer.lease_expires_at
            || root.role != Role::Shepherd
            || root.mode != "planning"
            || root.project_id != self.project_id
            || root.run != pending.run
            || root.harness != pending.expected_attachment.target
            || now < root.bound_at
            || now >= root.expires_at
        {
            return Err(denied());
        }
        match authority.load_review_custody(&engineer.agent_id) {
            Ok(custody) => validate_review_claim_custody(&custody)?,
            Err(DispatchStoreError::Io { source, .. })
                if source.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(error.into()),
        }
        let workspace = self.workspace_root.as_deref().ok_or_else(denied)?;
        let state = self.store.load_run(&pending.run)?;
        crate::orientation::verify_critic_pre(
            workspace,
            self.store.runs_root(),
            &state,
            engineer,
            engineer_pending,
        )
        .map_err(|error| {
            DispatchServiceError::InvalidRequest(
                error
                    .message_text()
                    .unwrap_or("native orientation pre is invalid")
                    .into(),
            )
        })
    }

    fn validate_review_replacement_claim(
        &self,
        pending: &PendingDispatch,
        root: &RootSessionBinding,
        now: i64,
        authority: &crate::dispatch_store::LockedDispatchRun<'_>,
    ) -> DispatchServiceResult<Option<crate::dispatch_store::ReviewTerminalSnapshot>> {
        let Some(subject) = pending.replaces_agent_id.as_ref() else {
            return Ok(None);
        };
        let source = authority.review_terminal_snapshot(subject)?;
        let original = &source.pending;
        let custody = &source.custody;
        let record = &source.subject;
        if &source.root != root
            || root.role != Role::Shepherd
            || root.project_id != self.project_id
            || root.run != pending.run
            || root.harness != pending.expected_attachment.target
            || now < root.bound_at
            || now >= root.expires_at
            || now < custody.updated_at
            || !root_replacement_phase(root, original, &pending.run_status)
            || pending.caller_role != Role::Shepherd
            || pending.parent_dispatch_id.is_some()
            || pending.expected_attachment.agent_id == *subject
            || pending.expected_child_session_id == original.expected_child_session_id
            || custody.state != ReviewCustodyState::Replaced
            || custody.replacement_agent_id.as_ref() != Some(&pending.expected_attachment.agent_id)
            || custody.project_id != pending.project_id
            || custody.run != pending.run
            || custody.root_session_id != root.session_id
            || custody.subject_agent_id != *subject
            || custody.subject_session_id != record.session_id
            || custody.subject_role != record.role
            || custody.lane != record.lane
            || custody.stopped_at != record.stopped_at
            || custody.task_sha256 != original.task_sha256
            || custody.pending_launch_id_hash != original.launch_id_hash
            || record.state != DispatchState::Malignant
            || record.project_id != pending.project_id
            || record.run != pending.run
            || record.root_session_id != root.session_id
            || record.role != original.role
            || record.lane != original.lane
            || record.harness != original.expected_attachment.target
            || record.agent_id != original.expected_attachment.agent_id
            || record.session_id != original.expected_child_session_id
            || record
                .write_scope
                .iter()
                .map(String::as_str)
                .collect::<Vec<_>>()
                != original
                    .write_scope
                    .iter()
                    .map(PathAuthority::as_str)
                    .collect::<Vec<_>>()
            || original.launch_state != PendingLaunchState::Quarantined
            || !same_replacement_contract(original, pending)
        {
            return Err(DispatchServiceError::InvalidRequest(
                "replacement claim requires exact current-root review-replace authorization and unchanged terminal lineage".into(),
            ));
        }
        Ok(Some(source))
    }

    fn authorize_dispatch_budget(
        &self,
        pending: &PendingDispatch,
        now: i64,
        authority: &crate::dispatch_store::LockedDispatchRun<'_>,
    ) -> DispatchServiceResult<()> {
        let workspace = self.workspace_root.as_deref().ok_or_else(|| {
            DispatchServiceError::InvalidRequest(
                "dispatch budget requires a bound workspace".into(),
            )
        })?;
        crate::dispatch_budget::authorize(
            workspace,
            &self.store,
            &self.project_id,
            pending,
            now,
            authority,
        )?;
        Ok(())
    }
}

fn store_domain_error(error: DispatchServiceError) -> DispatchStoreError {
    match error {
        DispatchServiceError::Store(error) => error,
        DispatchServiceError::Domain(error) => DispatchStoreError::Domain(error),
        other => DispatchStoreError::Domain(DispatchError::InvalidPending(other.to_string())),
    }
}

fn installed_package_binding_from_environment(
    workspace_root: &Path,
) -> DispatchServiceResult<InstalledPackageBinding> {
    let descriptor_path = std::env::var_os("SHEPHERD_NATIVE_DESCRIPTOR")
        .map(PathBuf::from)
        .filter(|path| path.is_absolute())
        .ok_or_else(|| {
            DispatchServiceError::InvalidRequest(
                "native installed-package descriptor is absent".into(),
            )
        })?;
    validate_descriptor_custody(&descriptor_path)?;
    let bytes = read_path_nofollow(&descriptor_path, MAX_ATTESTATION_BYTES)?;
    let descriptor: NativeTransportDescriptor =
        serde_json::from_slice(&bytes).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "native transport descriptor is malformed: {error}"
            ))
        })?;
    if descriptor.schema != "shepherd.native-transport/2"
        || descriptor.binary.is_empty()
        || !descriptor.env.is_object()
        || descriptor
            .auth_snapshot
            .as_deref()
            .is_some_and(str::is_empty)
        || digest_from_hex(&descriptor.candidate_sha256).is_err()
        || digest_from_hex(&descriptor.installed_manifest_sha256).is_err()
    {
        return Err(DispatchServiceError::InvalidRequest(
            "native transport descriptor identity is invalid".into(),
        ));
    }
    let bound_project = fs::canonicalize(&descriptor.project_root).map_err(|error| {
        DispatchServiceError::InvalidRequest(format!(
            "cannot resolve descriptor project root: {error}"
        ))
    })?;
    let current_project = fs::canonicalize(workspace_root).map_err(|error| {
        DispatchServiceError::InvalidRequest(format!("cannot resolve workspace root: {error}"))
    })?;
    if bound_project != current_project {
        return Err(DispatchServiceError::InvalidRequest(
            "native transport descriptor belongs to another project workspace".into(),
        ));
    }
    let binary = PathBuf::from(&descriptor.binary);
    let root = PathBuf::from(descriptor.installed_package_root);
    let manifest = PathBuf::from(descriptor.installed_manifest);
    if !binary.is_absolute()
        || !root.is_absolute()
        || !manifest.is_absolute()
        || manifest.file_name().and_then(|value| value.to_str()) != Some(".shepherd-generated.json")
    {
        return Err(DispatchServiceError::InvalidRequest(
            "native installed package paths must be absolute and canonical".into(),
        ));
    }
    reject_symlink_components(&binary)?;
    reject_symlink_components(&root)?;
    reject_symlink_components(&manifest)?;
    let binary = fs::canonicalize(binary).map_err(|error| {
        DispatchServiceError::InvalidRequest(format!(
            "cannot resolve native candidate binary: {error}"
        ))
    })?;
    let root = fs::canonicalize(root).map_err(|error| {
        DispatchServiceError::InvalidRequest(format!(
            "cannot resolve installed package root: {error}"
        ))
    })?;
    let manifest = fs::canonicalize(manifest).map_err(|error| {
        DispatchServiceError::InvalidRequest(format!(
            "cannot resolve installed package manifest: {error}"
        ))
    })?;
    if !manifest.starts_with(&root) || !manifest.is_file() {
        return Err(DispatchServiceError::InvalidRequest(
            "installed manifest is outside its retained package root".into(),
        ));
    }
    let current_binary = fs::canonicalize(std::env::current_exe().map_err(|error| {
        DispatchServiceError::InvalidRequest(format!(
            "cannot identify the executing Shepherd binary: {error}"
        ))
    })?)
    .map_err(|error| {
        DispatchServiceError::InvalidRequest(format!(
            "cannot canonicalize the executing Shepherd binary: {error}"
        ))
    })?;
    if binary != current_binary {
        return Err(DispatchServiceError::InvalidRequest(
            "native transport descriptor does not name the executing Shepherd binary".into(),
        ));
    }
    let candidate_sha256 =
        raw_file_sha256(&binary, MAX_NATIVE_BINARY_BYTES, FileLinks::SelfMeasured)?;
    if !constant_time_digest_eq(
        &candidate_sha256,
        &digest_from_hex(&descriptor.candidate_sha256)?,
    ) {
        return Err(DispatchServiceError::InvalidRequest(
            "native transport descriptor candidate digest does not match the executing binary"
                .into(),
        ));
    }
    let manifest_sha256 = raw_file_sha256(&manifest, MAX_ATTESTATION_BYTES, FileLinks::Unique)?;
    if !constant_time_digest_eq(
        &manifest_sha256,
        &digest_from_hex(&descriptor.installed_manifest_sha256)?,
    ) {
        return Err(DispatchServiceError::InvalidRequest(
            "native transport descriptor manifest digest does not match installed bytes".into(),
        ));
    }
    Ok(InstalledPackageBinding {
        binary,
        root_filesystem_id: filesystem_id(&root)?,
        root,
        manifest,
        candidate_sha256,
        manifest_sha256,
    })
}

fn local_installed_package_binding(
    root: &Path,
    manifest: &Path,
) -> DispatchServiceResult<InstalledPackageBinding> {
    reject_symlink_components(root)?;
    reject_symlink_components(manifest)?;
    let root = fs::canonicalize(root).map_err(|error| {
        DispatchServiceError::InvalidRequest(format!(
            "cannot resolve installed package root: {error}"
        ))
    })?;
    let manifest = fs::canonicalize(manifest).map_err(|error| {
        DispatchServiceError::InvalidRequest(format!(
            "cannot resolve installed package manifest: {error}"
        ))
    })?;
    if !manifest.starts_with(&root)
        || manifest.file_name().and_then(|value| value.to_str()) != Some(".shepherd-generated.json")
    {
        return Err(DispatchServiceError::InvalidRequest(
            "installed manifest is outside its retained package root".into(),
        ));
    }
    let binary = fs::canonicalize(std::env::current_exe().map_err(|error| {
        DispatchServiceError::InvalidRequest(format!(
            "cannot identify the executing Shepherd binary: {error}"
        ))
    })?)
    .map_err(|error| {
        DispatchServiceError::InvalidRequest(format!(
            "cannot canonicalize the executing Shepherd binary: {error}"
        ))
    })?;
    reject_symlink_components(&binary)?;
    Ok(InstalledPackageBinding {
        candidate_sha256: raw_file_sha256(
            &binary,
            MAX_NATIVE_BINARY_BYTES,
            FileLinks::SelfMeasured,
        )?,
        manifest_sha256: raw_file_sha256(&manifest, MAX_ATTESTATION_BYTES, FileLinks::Unique)?,
        root_filesystem_id: filesystem_id(&root)?,
        binary,
        root,
        manifest,
    })
}

fn validate_installed_package_binding(
    installed: &InstalledPackageBinding,
) -> DispatchServiceResult<()> {
    reject_symlink_components(&installed.binary)?;
    reject_symlink_components(&installed.root)?;
    reject_symlink_components(&installed.manifest)?;
    let current_binary = fs::canonicalize(std::env::current_exe().map_err(|error| {
        DispatchServiceError::InvalidRequest(format!(
            "cannot identify the executing Shepherd binary: {error}"
        ))
    })?)
    .map_err(|error| {
        DispatchServiceError::InvalidRequest(format!(
            "cannot canonicalize the executing Shepherd binary: {error}"
        ))
    })?;
    if current_binary != installed.binary
        || filesystem_id(&installed.root)? != installed.root_filesystem_id
        || !constant_time_digest_eq(
            &raw_file_sha256(
                &installed.binary,
                MAX_NATIVE_BINARY_BYTES,
                FileLinks::SelfMeasured,
            )?,
            &installed.candidate_sha256,
        )
        || !constant_time_digest_eq(
            &raw_file_sha256(
                &installed.manifest,
                MAX_ATTESTATION_BYTES,
                FileLinks::Unique,
            )?,
            &installed.manifest_sha256,
        )
    {
        return Err(DispatchError::AttachmentMismatch(
            "retained native candidate or installed package changed after bootstrap".into(),
        )
        .into());
    }
    Ok(())
}

fn validate_descriptor_custody(path: &Path) -> DispatchServiceResult<()> {
    reject_symlink_components(path)?;
    let metadata = fs::symlink_metadata(path).map_err(|error| {
        DispatchServiceError::InvalidRequest(format!(
            "cannot inspect native transport descriptor: {error}"
        ))
    })?;
    if metadata.file_type().is_symlink() || !metadata.is_file() {
        return Err(DispatchServiceError::InvalidRequest(
            "native transport descriptor must be a regular no-follow file".into(),
        ));
    }
    #[cfg(unix)]
    {
        use std::os::unix::fs::{MetadataExt, PermissionsExt};
        let owner = rustix::process::geteuid().as_raw();
        let parent = path.parent().ok_or_else(|| {
            DispatchServiceError::InvalidRequest(
                "native transport descriptor has no protected parent".into(),
            )
        })?;
        let parent_metadata = fs::symlink_metadata(parent).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot inspect native transport descriptor parent: {error}"
            ))
        })?;
        if metadata.nlink() != 1
            || metadata.uid() != owner
            || metadata.permissions().mode() & 0o077 != 0
            || parent_metadata.file_type().is_symlink()
            || !parent_metadata.is_dir()
            || parent_metadata.uid() != owner
            || parent_metadata.permissions().mode() & 0o022 != 0
        {
            return Err(DispatchServiceError::InvalidRequest(
                "native transport descriptor is not protected by owner-only custody".into(),
            ));
        }
    }
    #[cfg(windows)]
    if windows_file_identity(path)?.2 != 1 {
        return Err(DispatchServiceError::InvalidRequest(
            "native transport descriptor has multiple hard links".into(),
        ));
    }
    Ok(())
}

/// Whether a measured file must be the only name for its bytes.
///
/// The installed manifest must be: the Node consumer measures it with
/// `sha256UniqueFile`, so the two sides have to agree. The native binary must
/// not be, because cargo publishes `target/debug/<bin>` as a hard link into
/// `deps/` and demanding a single entry there refuses the standard toolchain
/// layout for the one file the caller does not choose.
#[derive(Clone, Copy, Eq, PartialEq)]
enum FileLinks {
    Unique,
    SelfMeasured,
}

fn raw_file_sha256(path: &Path, limit: usize, links: FileLinks) -> DispatchServiceResult<[u8; 32]> {
    #[cfg(unix)]
    {
        use rustix::fs::{FileType, fstat};
        let descriptor = open_path_nofollow(path)?;
        let stat = fstat(&descriptor).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot inspect no-follow digest source {}: {error}",
                path.display()
            ))
        })?;
        if !FileType::from_raw_mode(stat.st_mode).is_file()
            || (links == FileLinks::Unique && stat.st_nlink != 1)
            || stat.st_size < 0
            || usize::try_from(stat.st_size).map_or(true, |size| size > limit)
        {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "native digest source is not a bounded unique regular file: {}",
                path.display()
            )));
        }
        let mut file = File::from(descriptor);
        let mut digest = Sha256::new();
        let mut buffer = [0_u8; 64 * 1024];
        loop {
            let read = file.read(&mut buffer).map_err(|error| {
                DispatchServiceError::InvalidRequest(format!(
                    "cannot hash native file {}: {error}",
                    path.display()
                ))
            })?;
            if read == 0 {
                break;
            }
            digest.update(&buffer[..read]);
        }
        Ok(digest.finalize().into())
    }
    #[cfg(not(unix))]
    {
        // Mirror the Unix branch: uniqueness is the CALLER's policy. This used
        // to delegate to read_path_nofollow, which enforces uniqueness
        // unconditionally, so FileLinks::SelfMeasured was accepted on Unix and
        // silently ignored here -- and cargo hard-links target/debug/shepherd.exe
        // from deps/, so the self-measured binary always has two links.
        let bytes = crate::safe_fs::read_regular_nofollow(path, limit as u64).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot read no-follow digest source {}: {error}",
                path.display()
            ))
        })?;
        #[cfg(windows)]
        if links == FileLinks::Unique && windows_file_identity(path)?.2 != 1 {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "native digest source is not a bounded unique regular file: {}",
                path.display()
            )));
        }
        #[cfg(not(windows))]
        let _ = links;
        Ok(sha256(&bytes))
    }
}

fn selected_binding_run(
    store: &DispatchStore,
    requested: Option<&str>,
) -> DispatchServiceResult<RunId> {
    let Some(requested) = requested else {
        return Ok(store.resolve_active_run()?);
    };
    let run = RunId::new(requested)?;
    store.load_run(&run)?;
    Ok(run)
}

fn validate_root_binding_mode(mode: &str, status: &str) -> DispatchServiceResult<()> {
    let permitted = matches!(
        (mode, status),
        ("planning", "planted" | "planned") | ("execution", "executing")
    );
    if permitted {
        Ok(())
    } else {
        Err(DispatchServiceError::InvalidRequest(format!(
            "root mode `{mode}` is not permitted for run status `{status}`"
        )))
    }
}

fn validate_pending_schema(schema: &str) -> DispatchServiceResult<()> {
    if schema == PENDING_REQUEST_SCHEMA {
        Ok(())
    } else {
        Err(DispatchServiceError::InvalidRequest(format!(
            "unsupported pending dispatch schema `{schema}`"
        )))
    }
}

fn validate_profile_schema(schema: &str) -> DispatchServiceResult<()> {
    if schema == PROFILE_REQUEST_SCHEMA {
        Ok(())
    } else {
        Err(DispatchServiceError::InvalidRequest(format!(
            "unsupported profile schema `{schema}`"
        )))
    }
}

fn validate_skill_use_schema(schema: &str) -> DispatchServiceResult<()> {
    if schema == SKILL_USE_REQUEST_SCHEMA {
        Ok(())
    } else {
        Err(DispatchServiceError::InvalidRequest(format!(
            "unsupported skill-use schema `{schema}`"
        )))
    }
}

fn selected_broker_run(
    store: &DispatchStore,
    requested: &Option<String>,
    _parent: &crate::dispatch_broker::BrokerParent,
) -> DispatchServiceResult<RunId> {
    let Some(requested) = requested else {
        return Ok(store.resolve_active_run()?);
    };
    Ok(RunId::new(requested.clone())?)
}

fn parse_role(value: &str) -> DispatchServiceResult<Role> {
    Role::from_carrier(value)
        .or_else(|_| Role::from_name(value))
        .map_err(Into::into)
}

fn parse_work_kind(value: &str) -> DispatchServiceResult<WorkKind> {
    match value {
        "planning" => Ok(WorkKind::Planning),
        "production-code" => Ok(WorkKind::ProductionCode),
        "artifact" => Ok(WorkKind::Artifact),
        "review" => Ok(WorkKind::Review),
        "research" => Ok(WorkKind::Research),
        "coordination" => Ok(WorkKind::Coordination),
        _ => Err(DispatchServiceError::InvalidRequest(format!(
            "unknown pending work kind `{value}`"
        ))),
    }
}

fn validate_parent_dispatch(
    caller_dispatch: Option<&shepherd::dispatch::DispatchId>,
    requested: Option<&str>,
) -> DispatchServiceResult<()> {
    match (caller_dispatch, requested) {
        (None, None) => Ok(()),
        (Some(caller), Some(requested)) if caller.as_str() == requested => Ok(()),
        (Some(_), Some(_)) => Err(DispatchServiceError::InvalidRequest(
            "pending ancestry does not name the authenticated parent process".into(),
        )),
        _ => Err(DispatchServiceError::InvalidRequest(
            "root and child pending ancestry must be explicit and exact".into(),
        )),
    }
}

fn same_attachment(
    left: &CarrierAttachmentExpectation,
    right: &CarrierAttachmentExpectation,
) -> bool {
    left.target == right.target
        && left.role == right.role
        && left.agent_id == right.agent_id
        && left.installed_carrier_path == right.installed_carrier_path
        && constant_time_digest_eq(&left.candidate_sha256, &right.candidate_sha256)
        && constant_time_digest_eq(&left.carrier_sha256, &right.carrier_sha256)
        && constant_time_digest_eq(&left.compiler_tree_sha256, &right.compiler_tree_sha256)
        && left.startup_skill == right.startup_skill
        && constant_time_digest_eq(&left.skill_bundle_sha256, &right.skill_bundle_sha256)
        && left.attachment_kind == right.attachment_kind
}

fn attachment_kind_name(kind: AttachmentKind) -> &'static str {
    match kind {
        AttachmentKind::ClaudePreload => "claude-preload",
        AttachmentKind::CodexCustomAgent => "codex-custom-agent",
        AttachmentKind::PiSkillPath => "pi-skill-path",
    }
}

fn trusted_attachment_expectation(
    installed: &InstalledPackageBinding,
    request: CarrierAttachmentExpectationRequest,
) -> DispatchServiceResult<CarrierAttachmentExpectation> {
    let role = parse_role(&request.role)?;
    let agent_id = AgentId::new(request.agent_id)?;
    let attachment_kind = match request.attachment_kind.as_str() {
        "claude-preload" => AttachmentKind::ClaudePreload,
        "codex-custom-agent" => AttachmentKind::CodexCustomAgent,
        "pi-skill-path" => AttachmentKind::PiSkillPath,
        _ => {
            return Err(DispatchServiceError::InvalidRequest(
                "unknown carrier attachment kind".into(),
            ));
        }
    };
    attachment_kind.validate_for(request.target)?;
    let trusted = trusted_skill_carrier(installed, request.target, role, true)?;
    let expected = CarrierAttachmentExpectation {
        target: request.target,
        role,
        agent_id,
        installed_carrier_path: trusted.installed_carrier_path,
        candidate_sha256: trusted.candidate_sha256,
        carrier_sha256: trusted.carrier_sha256,
        compiler_tree_sha256: trusted.compiler_tree_sha256,
        startup_skill: trusted.startup_skill,
        skill_bundle_sha256: trusted.skill_bundle_sha256,
        attachment_kind,
    };
    expected.validate()?;
    Ok(expected)
}

/// Measure a compiler-owned role carrier without inventing a child identity
/// for a root or a temporary root profile.
fn trusted_skill_carrier(
    installed: &InstalledPackageBinding,
    target: Harness,
    role: Role,
    dispatchable: bool,
) -> DispatchServiceResult<TrustedSkillCarrier> {
    validate_installed_package_binding(installed)?;
    let root = installed.root.as_path();
    let manifest_path = installed.manifest.as_path();
    if !manifest_path.starts_with(root) {
        return Err(DispatchServiceError::InvalidRequest(
            "installed manifest escaped its retained package root".into(),
        ));
    }
    attachment_kind_for(target)?;
    let manifest_bytes = read_path_nofollow(manifest_path, MAX_ATTESTATION_BYTES)?;
    let manifest: serde_json::Value = serde_json::from_slice(&manifest_bytes).map_err(|error| {
        DispatchServiceError::Domain(DispatchError::AttachmentMismatch(format!(
            "trusted generated manifest is malformed: {error}"
        )))
    })?;
    if manifest.get("schema").and_then(serde_json::Value::as_str)
        != Some("shepherd.compiled-tree/4")
        || manifest.get("target").and_then(serde_json::Value::as_str) != Some(harness_name(target))
    {
        return Err(DispatchServiceError::Domain(
            DispatchError::AttachmentMismatch(
                "trusted generated manifest target or schema is invalid".into(),
            ),
        ));
    }
    let compiler_tree_sha256 = validate_compiled_tree_files(root, &manifest)?;
    let role_row = manifest
        .get("roles")
        .and_then(serde_json::Value::as_array)
        .and_then(|roles| {
            roles.iter().find(|candidate| {
                candidate.get("role").and_then(serde_json::Value::as_str) == Some(role.as_str())
            })
        })
        .ok_or_else(|| {
            DispatchServiceError::Domain(DispatchError::AttachmentMismatch(
                "trusted generated manifest has no requested role".into(),
            ))
        })?;
    if role_row.get("dispatchable") != Some(&serde_json::Value::Bool(dispatchable)) {
        return Err(DispatchServiceError::Domain(
            DispatchError::AttachmentMismatch(format!(
                "trusted generated role dispatchability is not `{dispatchable}`"
            )),
        ));
    }
    let carrier_relative = role_row
        .get("carrier_path")
        .and_then(serde_json::Value::as_str)
        .ok_or_else(|| {
            DispatchServiceError::Domain(DispatchError::AttachmentMismatch(
                "trusted generated role has no carrier path".into(),
            ))
        })?;
    let startup_skill = role_row
        .get("startup_skill")
        .and_then(serde_json::Value::as_str)
        .ok_or_else(|| {
            DispatchServiceError::Domain(DispatchError::AttachmentMismatch(
                "trusted generated role has no startup skill".into(),
            ))
        })?;
    let skill_digest = role_row
        .get("startup_skill_sha256")
        .and_then(serde_json::Value::as_str)
        .ok_or_else(|| {
            DispatchServiceError::Domain(DispatchError::AttachmentMismatch(
                "trusted generated role has no startup skill digest".into(),
            ))
        })?;
    let carrier_relative = trusted_manifest_relative_path(root, carrier_relative)?;
    // Codex compiles the two non-dispatchable root roles into its shared
    // config carrier. It is not a child custom-agent file.
    let carrier_kind = if target == Harness::Codex
        && !dispatchable
        && matches!(role, Role::Shepherd | Role::Planter)
        && carrier_relative == "shepherd.codex.toml"
    {
        "config"
    } else {
        "role"
    };
    validate_manifest_carrier_row(&manifest, &carrier_relative, carrier_kind)?;
    let carrier_path = root.join(&carrier_relative);
    let carrier_sha256 = hash_carrier_path(&carrier_path)?;
    let skill_path = unique_skill_path(root, &carrier_path, startup_skill, target)?;
    let skill_bundle_sha256 = hash_skill_bundle(&skill_path, startup_skill)?;
    let expected = TrustedSkillCarrier {
        installed_carrier_path: portable_absolute_path(&carrier_path),
        candidate_sha256: installed.candidate_sha256,
        carrier_sha256,
        compiler_tree_sha256,
        startup_skill: startup_skill.into(),
        skill_bundle_sha256,
    };
    if !constant_time_digest_eq(&digest_from_hex(skill_digest)?, &skill_bundle_sha256) {
        return Err(DispatchServiceError::Domain(
            DispatchError::AttachmentMismatch(
                "trusted startup skill digest does not match opened bytes".into(),
            ),
        ));
    }
    Ok(expected)
}

fn validate_compiled_tree_files(
    root: &Path,
    manifest: &serde_json::Value,
) -> DispatchServiceResult<[u8; 32]> {
    let expected_tree = manifest
        .get("tree_digest")
        .and_then(serde_json::Value::as_str)
        .ok_or_else(|| {
            DispatchServiceError::Domain(DispatchError::AttachmentMismatch(
                "trusted generated manifest has no tree digest".into(),
            ))
        })?;
    let expected_tree = digest_from_hex(expected_tree)?;
    let files = manifest
        .get("files")
        .and_then(serde_json::Value::as_array)
        .filter(|files| !files.is_empty() && files.len() <= MAX_ATTACHMENT_FILES)
        .ok_or_else(|| {
            DispatchServiceError::Domain(DispatchError::AttachmentMismatch(
                "trusted generated manifest has no bounded file inventory".into(),
            ))
        })?;
    let mut seen = BTreeSet::new();
    let mut total_bytes = 0_usize;
    let mut tree = Sha256::new();
    for row in files {
        let path = row
            .get("path")
            .and_then(serde_json::Value::as_str)
            .ok_or_else(|| attachment_mismatch("generated file row has no path"))?;
        let path = trusted_manifest_relative_path(root, path)?;
        if !seen.insert(path.clone()) {
            return Err(attachment_mismatch(
                "generated file inventory contains a duplicate path",
            ));
        }
        let kind = row
            .get("kind")
            .and_then(serde_json::Value::as_str)
            .filter(|kind| matches!(*kind, "role" | "skill" | "config"))
            .ok_or_else(|| attachment_mismatch("generated file row has an invalid kind"))?;
        let _ = kind;
        let mode = row
            .get("mode")
            .and_then(serde_json::Value::as_u64)
            .and_then(|value| u32::try_from(value).ok())
            .filter(|mode| matches!(*mode, 0o644 | 0o755))
            .ok_or_else(|| attachment_mismatch("generated file row has an invalid mode"))?;
        let expected_content = row
            .get("content_sha256")
            .and_then(serde_json::Value::as_str)
            .ok_or_else(|| attachment_mismatch("generated file row has no content digest"))?;
        let expected_content = digest_from_hex(expected_content)?;
        let expected_bytes = row
            .get("utf8_bytes")
            .and_then(serde_json::Value::as_u64)
            .and_then(|value| usize::try_from(value).ok())
            .ok_or_else(|| attachment_mismatch("generated file row has no byte length"))?;
        total_bytes = total_bytes.checked_add(expected_bytes).ok_or_else(|| {
            attachment_mismatch("generated file inventory byte length overflowed")
        })?;
        if total_bytes > MAX_ATTACHMENT_TREE_BYTES {
            return Err(attachment_mismatch(
                "generated file inventory exceeds the trusted tree byte limit",
            ));
        }
        let absolute = root.join(&path);
        let bytes = read_path_nofollow(&absolute, expected_bytes)?;
        if bytes.len() != expected_bytes
            || !constant_time_digest_eq(&sha256(&bytes), &expected_content)
        {
            return Err(attachment_mismatch(
                "installed file differs from its retained generated manifest",
            ));
        }
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let actual_mode = fs::symlink_metadata(&absolute)
                .map_err(|error| {
                    DispatchServiceError::InvalidRequest(format!(
                        "cannot inspect installed file mode {}: {error}",
                        absolute.display()
                    ))
                })?
                .permissions()
                .mode()
                & 0o777;
            if actual_mode != mode {
                return Err(attachment_mismatch(
                    "installed file mode differs from its retained generated manifest",
                ));
            }
        }
        tree.update(path.as_bytes());
        tree.update([0]);
        tree.update(mode.to_be_bytes());
        tree.update([0]);
        tree.update(&bytes);
        tree.update([0, 0]);
    }
    let actual_tree: [u8; 32] = tree.finalize().into();
    if !constant_time_digest_eq(&actual_tree, &expected_tree) {
        return Err(attachment_mismatch(
            "installed file inventory does not reproduce its compiler tree digest",
        ));
    }
    Ok(expected_tree)
}

fn validate_manifest_carrier_row(
    manifest: &serde_json::Value,
    carrier_relative: &str,
    expected_kind: &str,
) -> DispatchServiceResult<()> {
    let rows = manifest
        .get("files")
        .and_then(serde_json::Value::as_array)
        .ok_or_else(|| attachment_mismatch("generated manifest has no file inventory"))?;
    let matching = rows
        .iter()
        .filter(|row| row.get("path").and_then(serde_json::Value::as_str) == Some(carrier_relative))
        .collect::<Vec<_>>();
    if matching.len() != 1
        || matching[0].get("kind").and_then(serde_json::Value::as_str) != Some(expected_kind)
    {
        return Err(attachment_mismatch(
            "requested carrier has no unique compiler-owned file of the required kind",
        ));
    }
    Ok(())
}

fn attachment_mismatch(message: &str) -> DispatchServiceError {
    DispatchServiceError::Domain(DispatchError::AttachmentMismatch(message.into()))
}

fn root_replacement_phase(
    root: &RootSessionBinding,
    original: &PendingDispatch,
    status: &str,
) -> bool {
    match (root.mode.as_str(), status) {
        ("planning", "planted") => {
            original.run_status == "planted"
                && original.role == Role::Engineer
                && original.work_kind == WorkKind::Planning
                && original.lane.is_none()
                && original.caller_role == Role::Shepherd
                && original.parent_dispatch_id.is_none()
        }
        ("execution", "executing") => original.run_status == "executing",
        _ => false,
    }
}

pub(crate) fn same_replacement_contract(
    original: &PendingDispatch,
    replacement: &PendingDispatch,
) -> bool {
    original.project_id == replacement.project_id
        && original.project_filesystem_id == replacement.project_filesystem_id
        && original.run == replacement.run
        && original.run_status == replacement.run_status
        && original.root_session_id == replacement.root_session_id
        && original.role == replacement.role
        && original.work_kind == replacement.work_kind
        && original.lane == replacement.lane
        && original.baseline_commit == replacement.baseline_commit
        && original.read_scope == replacement.read_scope
        && original.write_scope == replacement.write_scope
        && original.result_artifact == replacement.result_artifact
        && original.review_artifact == replacement.review_artifact
        && original.task_path == replacement.task_path
        && original.task_sha256 == replacement.task_sha256
        && original.expected_attachment.target == replacement.expected_attachment.target
        && original.expected_attachment.role == replacement.expected_attachment.role
        && original.expected_attachment.installed_carrier_path
            == replacement.expected_attachment.installed_carrier_path
        && original.expected_attachment.candidate_sha256
            == replacement.expected_attachment.candidate_sha256
        && original.expected_attachment.carrier_sha256
            == replacement.expected_attachment.carrier_sha256
        && original.expected_attachment.compiler_tree_sha256
            == replacement.expected_attachment.compiler_tree_sha256
        && original.expected_attachment.startup_skill
            == replacement.expected_attachment.startup_skill
        && original.expected_attachment.skill_bundle_sha256
            == replacement.expected_attachment.skill_bundle_sha256
        && original.expected_attachment.attachment_kind
            == replacement.expected_attachment.attachment_kind
}

fn trusted_profile_attachment(
    installed: &InstalledPackageBinding,
    harness: Harness,
) -> DispatchServiceResult<ProfileAttachmentExpectation> {
    let trusted = trusted_skill_carrier(installed, harness, Role::Planter, false)?;
    if trusted.startup_skill != Profile::Planter.startup_skill() {
        return Err(DispatchError::AttachmentMismatch(
            "Planter profile is not attached to the exact `planting` startup bundle".into(),
        )
        .into());
    }
    let carrier_attachment_sha256 = profile_carrier_digest(&trusted);
    ProfileAttachmentExpectation::new(
        harness,
        Profile::Planter,
        trusted.startup_skill,
        trusted.candidate_sha256,
        carrier_attachment_sha256,
        crate::dispatch_broker::random_digest_32()
            .map_err(|error| DispatchServiceError::InvalidRequest(error.to_string()))?,
    )
    .map_err(Into::into)
}

fn profile_carrier_digest(trusted: &TrustedSkillCarrier) -> [u8; 32] {
    let mut bound = Vec::with_capacity(128);
    bound.extend_from_slice(&trusted.candidate_sha256);
    bound.extend_from_slice(&trusted.carrier_sha256);
    bound.extend_from_slice(&trusted.compiler_tree_sha256);
    bound.extend_from_slice(&trusted.skill_bundle_sha256);
    sha256(&bound)
}

fn attachment_kind_for(harness: Harness) -> DispatchServiceResult<AttachmentKind> {
    match harness {
        Harness::ClaudeCode => Ok(AttachmentKind::ClaudePreload),
        Harness::Codex => Ok(AttachmentKind::CodexCustomAgent),
        Harness::Pi => Ok(AttachmentKind::PiSkillPath),
        _ => Err(DispatchServiceError::InvalidRequest(
            "profile and skill-use authority supports only Claude, Codex, and Pi".into(),
        )),
    }
}

fn validate_record_startup_attachment(
    record: &DispatchRecord,
    trusted: &TrustedSkillCarrier,
) -> DispatchServiceResult<()> {
    let startup = record.startup_attachment.as_ref().ok_or_else(|| {
        DispatchServiceError::Domain(DispatchError::AttachmentMismatch(
            "active dispatch has no Native startup attachment".into(),
        ))
    })?;
    if startup.skill != trusted.startup_skill
        || digest_from_hex(&startup.bundle_digest)? != trusted.skill_bundle_sha256
    {
        return Err(DispatchError::AttachmentMismatch(
            "active dispatch startup attachment differs from installed trusted bytes".into(),
        )
        .into());
    }
    Ok(())
}

fn validate_skill_use_record(
    record: &DispatchRecord,
    harness: Harness,
    session_id: &SessionId,
    now: i64,
) -> DispatchServiceResult<()> {
    record.validate_loaded()?;
    if record.harness != harness
        || &record.session_id != session_id
        || record.state != DispatchState::Active
        || now < record.started_at
        || now >= record.lease_expires_at
    {
        return Err(DispatchServiceError::InvalidRequest(
            "skill-use requires the exact live dispatch session and carrier".into(),
        ));
    }
    Ok(())
}

fn validate_profile_root(
    root: &RootSessionBinding,
    harness: Harness,
    workspace_root: Option<&Path>,
) -> DispatchServiceResult<()> {
    validate_root_binding_filesystem(root, workspace_root)?;
    if root.harness != harness || root.role != Role::Shepherd || root.mode != "planning" {
        return Err(DispatchServiceError::InvalidRequest(
            "profile transition requires the exact bound planning root and carrier".into(),
        ));
    }
    Ok(())
}

fn unique_skill_path(
    package_root: &Path,
    carrier: &Path,
    skill: &str,
    target: Harness,
) -> DispatchServiceResult<PathBuf> {
    // `installed_carrier_path` is retained in the PORTABLE spelling -- see
    // `portable_absolute_path`: forward slashes, verbatim `\\?\C:` unwrapped --
    // while `package_root` is a native path. On Unix the two spellings are
    // byte-identical, so `Path::starts_with` held. On Windows one side is
    // `C:/x` and the other `C:\x` (or `\\?\C:\x`), so containment never held
    // and every trusted carrier was reported as escaping its package root.
    // Compare in one spelling; the trailing separator keeps `/a/b` from being
    // read as containing `/a/bc`, which is what Path::starts_with gave us.
    let root_text = portable_absolute_path(package_root);
    let carrier_text = portable_absolute_path(carrier);
    let contained = carrier_text == root_text
        || carrier_text
            .strip_prefix(&root_text)
            .is_some_and(|rest| rest.starts_with('/'));
    if !contained {
        return Err(DispatchError::AttachmentMismatch(
            "trusted carrier escaped its retained installed package root".into(),
        )
        .into());
    }
    reject_symlink_components(package_root)?;
    reject_symlink_components(carrier)?;
    let relative = if target == Harness::Codex {
        Path::new(".agents/skills").join(skill)
    } else {
        Path::new("skills").join(skill)
    };
    let candidate = package_root.join(relative);
    reject_symlink_components(&candidate)?;
    match fs::symlink_metadata(&candidate) {
        Ok(metadata) if metadata.file_type().is_symlink() => Err(
            DispatchError::AttachmentMismatch("trusted installed skill path is a symlink".into())
                .into(),
        ),
        Ok(metadata) if metadata.is_dir() || metadata.is_file() => Ok(candidate),
        Ok(_) => Err(DispatchError::AttachmentMismatch(
            "trusted installed skill path is not a regular bundle".into(),
        )
        .into()),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Err(
            DispatchError::AttachmentMismatch("trusted installed skill bundle is absent".into())
                .into(),
        ),
        Err(error) => Err(DispatchServiceError::InvalidRequest(error.to_string())),
    }
}

fn digest_from_hex(value: &str) -> DispatchServiceResult<[u8; 32]> {
    if value.len() != 64 {
        return Err(DispatchServiceError::InvalidRequest(
            "digest must contain exactly 64 hexadecimal characters".into(),
        ));
    }
    let mut digest = [0; 32];
    let (pairs, remainder) = value.as_bytes().as_chunks::<2>();
    if !remainder.is_empty() {
        return Err(DispatchServiceError::InvalidRequest(
            "digest must contain pairs of hexadecimal characters".into(),
        ));
    }
    for (index, pair) in pairs.iter().enumerate() {
        let high = hex_digit(pair[0]).ok_or_else(|| {
            DispatchServiceError::InvalidRequest("digest contains a non-hex character".into())
        })?;
        let low = hex_digit(pair[1]).ok_or_else(|| {
            DispatchServiceError::InvalidRequest("digest contains a non-hex character".into())
        })?;
        digest[index] = (high << 4) | low;
    }
    Ok(digest)
}

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),
        b'A'..=b'F' => Some(value - b'A' + 10),
        _ => None,
    }
}

fn sha256(bytes: &[u8]) -> [u8; 32] {
    Sha256::digest(bytes).into()
}

fn hex(bytes: &[u8; 32]) -> String {
    let mut output = String::with_capacity(64);
    for byte in bytes {
        output.push_str(&format!("{byte:02x}"));
    }
    output
}

fn current_git_commit(root: Option<&Path>) -> DispatchServiceResult<GitCommit> {
    let root = root.ok_or_else(|| {
        DispatchServiceError::InvalidRequest("pending dispatch requires a repository root".into())
    })?;
    let git = trusted_git_executable()?;
    let output = Command::new(git)
        .env_clear()
        .args(["rev-parse", "HEAD"])
        .current_dir(root)
        .output()
        .map_err(|error| {
            DispatchServiceError::InvalidRequest(format!("cannot resolve git HEAD: {error}"))
        })?;
    if !output.status.success() {
        return Err(DispatchServiceError::InvalidRequest(
            "native repository has no resolvable HEAD".into(),
        ));
    }
    let value = String::from_utf8(output.stdout)
        .map_err(|_| DispatchServiceError::InvalidRequest("git HEAD is not UTF-8".into()))?;
    Ok(GitCommit::new(value.trim())?)
}

pub(crate) fn trusted_git_executable() -> DispatchServiceResult<PathBuf> {
    #[cfg(unix)]
    let candidates = [
        "/usr/bin/git",
        "/usr/local/bin/git",
        "/opt/homebrew/bin/git",
        "/opt/local/bin/git",
    ];
    #[cfg(windows)]
    let candidates = [
        r"C:\Program Files\Git\cmd\git.exe",
        r"C:\Program Files\Git\bin\git.exe",
    ];
    #[cfg(not(any(unix, windows)))]
    let candidates: [&str; 0] = [];
    candidates
        .iter()
        .map(Path::new)
        .find(|path| {
            fs::symlink_metadata(path).ok().is_some_and(|metadata| {
                metadata.is_file()
                    && reject_symlink_components(path).is_ok()
                    && fs::canonicalize(path).is_ok()
            })
        })
        .map(Path::to_path_buf)
        .ok_or_else(|| {
            DispatchServiceError::InvalidRequest(
                "no trusted absolute Git implementation is installed".into(),
            )
        })
}

fn root_binding_filesystem_id(root: Option<&Path>) -> DispatchServiceResult<Option<String>> {
    let root = root.ok_or_else(|| {
        DispatchServiceError::InvalidRequest(
            "root binding requires the current workspace root".into(),
        )
    })?;
    filesystem_id(root).map(|identity| Some(identity.to_string()))
}

fn validate_root_binding_filesystem(
    binding: &RootSessionBinding,
    workspace_root: Option<&Path>,
) -> DispatchServiceResult<()> {
    let current = root_binding_filesystem_id(workspace_root)?;
    let legacy_match = binding
        .project_filesystem_id
        .as_deref()
        .zip(workspace_root)
        .map(|(expected, root)| legacy_root_filesystem_matches(expected, root))
        .transpose()?
        .unwrap_or(false);
    if binding.project_filesystem_id != current && !legacy_match {
        return Err(DispatchServiceError::InvalidRequest(
            "root session binding belongs to a different workspace filesystem identity".into(),
        ));
    }
    Ok(())
}

fn legacy_root_filesystem_matches(expected: &str, root: &Path) -> DispatchServiceResult<bool> {
    #[cfg(unix)]
    {
        use rustix::fs::{Mode, OFlags, open};
        use std::os::unix::fs::MetadataExt;
        let Some(raw) = expected.strip_prefix("unix:") else {
            return Ok(false);
        };
        let Some((device, inode)) = raw.split_once(':') else {
            return Ok(false);
        };
        let (Ok(device), Ok(inode)) = (
            u64::from_str_radix(device, 16),
            u64::from_str_radix(inode, 16),
        ) else {
            return Ok(false);
        };
        // The ordinary hashed identity check above already verified canonical
        // path confinement; this no-follow descriptor measures the legacy pair.
        let descriptor = open(
            root,
            OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::empty(),
        )
        .map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot anchor legacy workspace root: {error}"
            ))
        })?;
        let metadata = File::from(descriptor).metadata().map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot inspect legacy workspace root: {error}"
            ))
        })?;
        Ok(metadata.dev() == device && metadata.ino() == inode)
    }
    #[cfg(not(unix))]
    {
        let _ = (expected, root);
        Ok(false)
    }
}

fn filesystem_id(root: &Path) -> DispatchServiceResult<ProjectFilesystemId> {
    let canonical = fs::canonicalize(root).map_err(|error| {
        DispatchServiceError::InvalidRequest(format!("cannot resolve project filesystem: {error}"))
    })?;
    if canonical != root {
        return Err(DispatchServiceError::InvalidRequest(
            "workspace root is not canonical".into(),
        ));
    }
    let metadata = fs::symlink_metadata(root).map_err(|error| {
        DispatchServiceError::InvalidRequest(format!("cannot inspect workspace root: {error}"))
    })?;
    if metadata.file_type().is_symlink() || !metadata.is_dir() {
        return Err(DispatchServiceError::InvalidRequest(
            "workspace root must be a regular directory, not a symlink".into(),
        ));
    }
    #[cfg(unix)]
    let material = {
        use rustix::fs::{FileType, Mode, OFlags, open};
        let descriptor = open(
            root,
            OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::empty(),
        )
        .map_err(|error| {
            DispatchServiceError::InvalidRequest(format!("cannot anchor workspace root: {error}"))
        })?;
        let file = File::from(descriptor);
        #[cfg(unix)]
        use std::os::unix::fs::MetadataExt;
        let stat = file.metadata().map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot inspect workspace root identity: {error}"
            ))
        })?;
        if !FileType::from_raw_mode(
            rustix::fs::fstat(&file)
                .map_err(|error| {
                    DispatchServiceError::InvalidRequest(format!(
                        "cannot stat workspace root: {error}"
                    ))
                })?
                .st_mode,
        )
        .is_dir()
        {
            return Err(DispatchServiceError::InvalidRequest(
                "workspace root descriptor is not a directory".into(),
            ));
        }
        format!("{}\0{}\0{}", canonical.display(), stat.dev(), stat.ino())
    };
    #[cfg(windows)]
    let material = {
        let (volume, file_id, links) = windows_file_identity(root)?;
        if links == 0 {
            return Err(DispatchServiceError::InvalidRequest(
                "workspace root has invalid Windows link metadata".into(),
            ));
        }
        format!(
            "{}\0{}\0{}\0{}",
            portable_absolute_path(&canonical),
            volume,
            file_id,
            links
        )
    };
    #[cfg(not(any(unix, windows)))]
    let material = format!("{}\0{}", canonical.display(), metadata.len());
    Ok(ProjectFilesystemId::new(hex(&sha256(material.as_bytes())))?)
}

fn repository_relative_path(root: Option<&Path>, candidate: &str) -> DispatchServiceResult<String> {
    let root = root.ok_or_else(|| {
        DispatchServiceError::InvalidRequest("pending dispatch requires a repository root".into())
    })?;
    let relative = normalized_relative_path(root, candidate)?;
    check_existing_path_case(root, &relative)?;
    Ok(relative)
}

fn trusted_manifest_relative_path(root: &Path, candidate: &str) -> DispatchServiceResult<String> {
    if Path::new(candidate).is_absolute() {
        return Err(DispatchServiceError::InvalidRequest(
            "generated manifest path must be repository-relative".into(),
        ));
    }
    let relative = normalized_relative_path(root, candidate)?;
    if !check_existing_path_case(root, &relative)? {
        return Err(DispatchServiceError::InvalidRequest(
            "generated manifest path is absent from the retained installed package".into(),
        ));
    }
    Ok(relative)
}

/// Preserve existing directory-entry spelling without treating case-folded
/// aliases as new paths. `false` means an unaliased component is absent, which
/// is valid for proposed repository writes but never for an installed carrier.
fn check_existing_path_case(root: &Path, relative: &str) -> DispatchServiceResult<bool> {
    let mut parent = root.to_path_buf();
    for component in relative.split('/') {
        let entries = fs::read_dir(&parent).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot inspect repository path parent: {error}"
            ))
        })?;
        let mut exact = false;
        let mut case_alias = false;
        for entry in entries {
            let name = entry
                .map_err(|error| {
                    DispatchServiceError::InvalidRequest(format!(
                        "cannot inspect repository directory entry: {error}"
                    ))
                })?
                .file_name();
            exact |= name == std::ffi::OsStr::new(component);
            case_alias |= name
                .to_str()
                .is_some_and(|name| name.eq_ignore_ascii_case(component));
        }
        if !exact {
            if case_alias {
                return Err(DispatchServiceError::InvalidRequest(
                    "repository path does not match exact existing directory-entry case".into(),
                ));
            }
            return Ok(false);
        }
        parent.push(component);
    }
    Ok(true)
}

fn normalized_relative_path(root: &Path, candidate: &str) -> DispatchServiceResult<String> {
    if candidate.is_empty()
        || candidate.len() > 4_096
        || candidate.contains(['\\', '\0', ':'])
        || candidate.chars().any(char::is_control)
        || !candidate.is_ascii()
        || candidate.contains("//")
        || candidate.starts_with("./")
        || candidate.contains("/./")
        || candidate.ends_with('/')
    {
        return Err(DispatchServiceError::InvalidRequest(
            "repository path is unsafe".into(),
        ));
    }
    let candidate_path = Path::new(candidate);
    let relative = if candidate_path.is_absolute() {
        candidate_path
            .strip_prefix(root)
            .map_err(|_| {
                DispatchServiceError::InvalidRequest("repository path escapes root".into())
            })?
            .to_path_buf()
    } else {
        candidate_path.to_path_buf()
    };
    let mut parts = Vec::new();
    for component in relative.components() {
        match component {
            std::path::Component::Normal(part) => {
                let part = part.to_str().ok_or_else(|| {
                    DispatchServiceError::InvalidRequest("repository path must be UTF-8".into())
                })?;
                if part.is_empty()
                    || part.ends_with('.')
                    || part.ends_with(' ')
                    || part.contains('~')
                {
                    return Err(DispatchServiceError::InvalidRequest(
                        "repository path uses an ambiguous alias form".into(),
                    ));
                }
                parts.push(part);
            }
            _ => {
                return Err(DispatchServiceError::InvalidRequest(
                    "repository path must be normalized and relative".into(),
                ));
            }
        }
    }
    if parts.is_empty() {
        return Err(DispatchServiceError::InvalidRequest(
            "repository path cannot name the root".into(),
        ));
    }
    Ok(parts.join("/"))
}

fn read_relative_nofollow(
    root: Option<&Path>,
    relative: &str,
    limit: usize,
) -> DispatchServiceResult<Vec<u8>> {
    let root = root.ok_or_else(|| {
        DispatchServiceError::InvalidRequest("pending dispatch requires a repository root".into())
    })?;
    let path = root.join(relative);
    read_path_nofollow(&path, limit)
}

fn read_path_nofollow(path: &Path, limit: usize) -> DispatchServiceResult<Vec<u8>> {
    #[cfg(unix)]
    {
        use rustix::fs::{FileType, fstat};
        let descriptor = open_path_nofollow(path)?;
        let stat = fstat(&descriptor).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot stat no-follow file {}: {error}",
                path.display()
            ))
        })?;
        if !FileType::from_raw_mode(stat.st_mode).is_file() {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "pending file is not a regular file: {}",
                path.display()
            )));
        }
        if stat.st_nlink != 1 {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "pending file has multiple hard links: {}",
                path.display()
            )));
        }
        read_bounded_file(File::from(descriptor), limit)
    }
    #[cfg(not(unix))]
    {
        let bytes = crate::safe_fs::read_regular_nofollow(path, limit as u64).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot read no-follow file {}: {error}",
                path.display()
            ))
        })?;
        #[cfg(windows)]
        if windows_file_identity(path)?.2 != 1 {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "pending file has multiple hard links: {}",
                path.display()
            )));
        }
        Ok(bytes)
    }
}

#[cfg(unix)]
fn read_bounded_file(file: File, limit: usize) -> DispatchServiceResult<Vec<u8>> {
    let mut bytes = Vec::new();
    file.take(u64::try_from(limit.saturating_add(1)).expect("bounded read fits u64"))
        .read_to_end(&mut bytes)
        .map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot read bounded native file: {error}"
            ))
        })?;
    if bytes.len() > limit {
        return Err(DispatchServiceError::InvalidRequest(format!(
            "native file exceeds {limit}-byte limit"
        )));
    }
    Ok(bytes)
}

#[cfg(unix)]
fn open_path_nofollow(path: &Path) -> DispatchServiceResult<rustix::fd::OwnedFd> {
    use rustix::fs::{Mode, OFlags, open, openat};
    if !path.is_absolute() {
        return Err(DispatchServiceError::InvalidRequest(
            "native no-follow path must be absolute".into(),
        ));
    }
    let mut descriptor = open(
        "/",
        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
        Mode::empty(),
    )
    .map_err(|error| {
        DispatchServiceError::InvalidRequest(format!("cannot open filesystem root: {error}"))
    })?;
    let components = path.components().collect::<Vec<_>>();
    for (index, component) in components.iter().enumerate() {
        let std::path::Component::Normal(name) = component else {
            if matches!(component, std::path::Component::RootDir) {
                continue;
            }
            return Err(DispatchServiceError::InvalidRequest(
                "native no-follow path contains an unsafe component".into(),
            ));
        };
        let last = index + 1 == components.len();
        let flags = OFlags::RDONLY
            | OFlags::CLOEXEC
            | OFlags::NOFOLLOW
            | OFlags::NONBLOCK
            | if last {
                OFlags::empty()
            } else {
                OFlags::DIRECTORY
            };
        descriptor = openat(&descriptor, *name, flags, Mode::empty()).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot open no-follow path {}: {error}",
                path.display()
            ))
        })?;
    }
    Ok(descriptor)
}

fn reject_symlink_components(path: &Path) -> DispatchServiceResult<()> {
    let mut current = PathBuf::new();
    for component in path.components() {
        if matches!(
            component,
            std::path::Component::Prefix(_) | std::path::Component::RootDir
        ) {
            current.push(component.as_os_str());
            continue;
        }
        current.push(component.as_os_str());
        match fs::symlink_metadata(&current) {
            Ok(metadata) if metadata.file_type().is_symlink() => {
                return Err(DispatchServiceError::InvalidRequest(format!(
                    "refusing symlink component {}",
                    current.display()
                )));
            }
            Ok(_) => {}
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => break,
            Err(error) => {
                return Err(DispatchServiceError::InvalidRequest(format!(
                    "cannot inspect pending path {}: {error}",
                    current.display()
                )));
            }
        }
    }
    Ok(())
}

struct NativeScopePolicy<'a> {
    run: &'a RunId,
    role: Role,
    work_kind: WorkKind,
    lane: Option<&'a LaneId>,
    result_artifact: &'a PathAuthority,
}

fn validate_native_scopes(
    root: Option<&Path>,
    policy: NativeScopePolicy<'_>,
    read_scope: &[PathAuthority],
    write_scope: &[PathAuthority],
) -> DispatchServiceResult<()> {
    let NativeScopePolicy {
        run,
        role,
        work_kind,
        lane,
        result_artifact,
    } = policy;
    if read_scope.is_empty() {
        return Err(DispatchServiceError::InvalidRequest(
            "pending read scope cannot be empty".into(),
        ));
    }
    for scope in read_scope.iter().chain(write_scope) {
        validate_scope_nofollow(root, scope)?;
        let class = classify_scope(root, scope, run)?;
        if write_scope.iter().any(|candidate| candidate == scope)
            && (!role.may_write_class(class)
                || !role_work_kind_allows(role, work_kind)
                || !role_scoped_path_allows(role, class, scope, run, lane, result_artifact))
        {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "native role `{role}` has no {:?} authority for `{scope}` under `{work_kind:?}`",
                class
            )));
        }
    }
    if matches!(
        role,
        Role::Coder | Role::Worker | Role::Engineer | Role::Conductor
    ) && write_scope.is_empty()
    {
        return Err(DispatchServiceError::InvalidRequest(format!(
            "native role `{role}` requires a bounded write scope"
        )));
    }
    if matches!(role, Role::Auditor | Role::Critic | Role::Discovery) && !write_scope.is_empty() {
        return Err(DispatchServiceError::InvalidRequest(format!(
            "native read/review role `{role}` cannot receive production write scope"
        )));
    }
    Ok(())
}

fn classify_scope(
    root: Option<&Path>,
    scope: &PathAuthority,
    run: &RunId,
) -> DispatchServiceResult<PathClass> {
    let raw = scope.as_str();
    let literal = raw.split('*').next().unwrap_or(raw).trim_end_matches('/');
    if literal.is_empty() {
        return Err(DispatchServiceError::InvalidRequest(
            "scope is too broad to classify natively".into(),
        ));
    }
    let root = root.ok_or_else(|| {
        DispatchServiceError::InvalidRequest(
            "native path classification requires the bound workspace root".into(),
        )
    })?;
    let run_prefix = format!(".shepherd/runs/{}/", run.as_str());
    let run_relative = literal.strip_prefix(&run_prefix);
    let native_state = literal == ".shepherd/project.json"
        || literal == ".shepherd/shepherd.db"
        || run_relative.is_some_and(|value| {
            value == "run.json"
                || value.starts_with("dispatch/")
                || value.starts_with("locks/")
                || value.starts_with("skill-use/")
                || value.starts_with("profile/")
        });
    let run_artifact = run_relative.is_some_and(is_run_artifact_path);
    let first = literal.split('/').next().unwrap_or_default();
    let production = [
        "crates", "packages", "content", "services", "scripts", "hooks", ".github", ".cargo",
    ];
    let components = literal.split('/').collect::<Vec<_>>();
    let known_production = production.contains(&first)
        || matches!(
            first,
            "Cargo.toml" | "Cargo.lock" | "package.json" | "package-lock.json" | "bun.lock"
        )
        || components.iter().any(|component| {
            matches!(
                *component,
                "src"
                    | "test"
                    | "tests"
                    | "build"
                    | "config"
                    | "migrations"
                    | "workflows"
                    | "hooks"
                    | "scripts"
            )
        });
    let explicit_non_code = [
        "README.md",
        "REPORT.md",
        "CHANGELOG.md",
        "SECURITY.md",
        "CONTRIBUTING.md",
        "CODE_OF_CONDUCT.md",
        "LICENSE",
    ];
    let known_non_code =
        literal == "docs" || literal.starts_with("docs/") || explicit_non_code.contains(&literal);
    let metadata = fs::symlink_metadata(root.join(literal));
    let kind = match metadata {
        Ok(metadata) if metadata.is_file() => PathFactKind::ExistingFile,
        Ok(metadata) if metadata.is_dir() => PathFactKind::ExistingDirectory,
        Ok(_) => {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "scope `{scope}` is not a regular file or directory"
            )));
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => PathFactKind::Missing,
        Err(error) => {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "cannot classify scope `{scope}`: {error}"
            )));
        }
    };
    let inherited_parent = if kind == PathFactKind::Missing {
        if known_non_code {
            Some(PathClass::NonCodeDeliverable)
        } else if run_artifact {
            Some(PathClass::RunArtifact)
        } else {
            None
        }
    } else {
        None
    };
    let git_tracked = git_tracks_path(root, literal)?;
    let facts = TrustedPathFacts {
        path: scope.clone(),
        kind,
        git_tracked,
        workspace_member: production.contains(&first),
        known_production,
        known_non_code,
        run_artifact,
        native_state,
        inherited_parent,
        conflict: [known_production, known_non_code, run_artifact, native_state]
            .into_iter()
            .filter(|value| *value)
            .count()
            > 1,
    };
    Ok(classify_trusted_path(&facts))
}

fn is_run_artifact_path(value: &str) -> bool {
    matches!(
        value,
        "mesh.md" | "seed.md" | "phase0.md" | "plan.md" | "handoff.md" | "close.md"
    ) || ["graph/", "lanes/", "reports/", "reviews/", "evidence/"]
        .iter()
        .any(|prefix| value.starts_with(prefix))
}

fn role_work_kind_allows(role: Role, work_kind: WorkKind) -> bool {
    match role {
        Role::Engineer => work_kind == WorkKind::Planning,
        Role::Conductor => work_kind == WorkKind::Coordination,
        Role::Coder => work_kind == WorkKind::ProductionCode,
        Role::Worker => work_kind == WorkKind::Artifact,
        Role::Auditor | Role::Critic | Role::Discovery | Role::Planter | Role::Shepherd => false,
    }
}

fn role_scoped_path_allows(
    role: Role,
    class: PathClass,
    scope: &PathAuthority,
    run: &RunId,
    lane: Option<&LaneId>,
    result_artifact: &PathAuthority,
) -> bool {
    if role == Role::Worker {
        return scope.is_exact()
            && match class {
                PathClass::NonCodeDeliverable => true,
                PathClass::RunArtifact => scope == result_artifact,
                PathClass::Production | PathClass::NativeState => false,
            };
    }
    if !matches!(role, Role::Engineer | Role::Conductor) {
        return true;
    }
    if class != PathClass::RunArtifact || !scope.is_exact() {
        return false;
    }
    let prefix = format!(".shepherd/runs/{run}/");
    let relative = scope.as_str().strip_prefix(&prefix).unwrap_or_default();
    match role {
        Role::Engineer => {
            matches!(relative, "phase0.md" | "plan.md" | "graph/topology.json")
                || engineer_lane_plan(relative)
                || engineer_planning_report(relative)
        }
        Role::Conductor => {
            let Some(lane) = lane else { return false };
            let Some(relative) = relative.strip_prefix("lanes/") else {
                return false;
            };
            let Some(relative) = relative.strip_prefix(lane.as_str()) else {
                return false;
            };
            let Some(relative) = relative.strip_prefix('/') else {
                return false;
            };
            relative == "handoff.md"
                || ["reports/", "reviews/", "evidence/"].iter().any(|prefix| {
                    relative.starts_with(prefix) && safe_artifact_tail(&relative[prefix.len()..])
                })
        }
        _ => true,
    }
}

fn validate_role_artifact_namespaces(
    run: &RunId,
    role: Role,
    lane: Option<&LaneId>,
    result_artifact: &PathAuthority,
    review_artifact: &PathAuthority,
) -> DispatchServiceResult<()> {
    if role != Role::Worker {
        return Ok(());
    }
    let lane = lane.ok_or_else(|| {
        DispatchServiceError::InvalidRequest(
            "Worker artifact custody requires its verified Conductor lane".into(),
        )
    })?;
    let result_prefix = format!(".shepherd/runs/{run}/lanes/{lane}/workers/");
    let review_prefix = format!(".shepherd/runs/{run}/lanes/{lane}/reviews/");
    let result_tail = result_artifact
        .as_str()
        .strip_prefix(&result_prefix)
        .filter(|tail| safe_artifact_tail(tail));
    let review_tail = review_artifact
        .as_str()
        .strip_prefix(&review_prefix)
        .filter(|tail| safe_artifact_tail(tail));
    if !result_artifact.is_exact()
        || !review_artifact.is_exact()
        || result_tail.is_none()
        || review_tail.is_none()
    {
        return Err(DispatchServiceError::InvalidRequest(
            "Worker result and review artifacts must use the exact lane-bound workers/reviews namespaces"
                .into(),
        ));
    }
    Ok(())
}

fn engineer_lane_plan(relative: &str) -> bool {
    let Some(relative) = relative.strip_prefix("lanes/") else {
        return false;
    };
    let Some((lane, tail)) = relative.split_once('/') else {
        return false;
    };
    !lane.is_empty()
        && lane
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
        && tail == "plan.md"
}

fn engineer_planning_report(relative: &str) -> bool {
    relative
        .strip_prefix("reports/planning-")
        .is_some_and(|tail| safe_artifact_tail(tail) && tail.ends_with(".md"))
}

fn safe_artifact_tail(value: &str) -> bool {
    !value.is_empty()
        && value.len() <= 255
        && !value.contains('/')
        && value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'-' | b'_'))
        && value != "."
        && value != ".."
}

fn git_tracks_path(root: &Path, path: &str) -> DispatchServiceResult<bool> {
    let git = trusted_git_executable()?;
    #[cfg(windows)]
    let null_config = "NUL";
    #[cfg(not(windows))]
    let null_config = "/dev/null";
    let output = Command::new(git)
        .env_clear()
        .env("GIT_CONFIG_NOSYSTEM", "1")
        .env("GIT_CONFIG_GLOBAL", null_config)
        .args(["-c", "core.fsmonitor=false", "ls-files", "--", path])
        .current_dir(root)
        .output()
        .map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot query trusted Git index for `{path}`: {error}"
            ))
        })?;
    if !output.status.success() {
        return Err(DispatchServiceError::InvalidRequest(format!(
            "trusted Git index query failed for `{path}`"
        )));
    }
    Ok(!output.stdout.is_empty())
}

fn validate_scope_nofollow(
    root: Option<&Path>,
    scope: &PathAuthority,
) -> DispatchServiceResult<()> {
    let Some(root) = root else {
        return Err(DispatchServiceError::InvalidRequest(
            "native path authority requires the bound workspace root".into(),
        ));
    };
    let prefix = scope
        .as_str()
        .split('*')
        .next()
        .unwrap_or(scope.as_str())
        .trim_end_matches('/');
    if prefix.is_empty() {
        return Err(DispatchServiceError::InvalidRequest(
            "native path authority cannot derive a safe prefix".into(),
        ));
    }
    let path = root.join(prefix);
    reject_symlink_components(&path)?;
    check_existing_path_case(root, prefix)?;
    match fs::symlink_metadata(&path) {
        Ok(metadata) if metadata.file_type().is_symlink() => Err(
            DispatchServiceError::InvalidRequest(format!("scope `{scope}` is a symlink")),
        ),
        Ok(metadata) if metadata.is_file() || metadata.is_dir() => Ok(()),
        Ok(_) => Err(DispatchServiceError::InvalidRequest(format!(
            "scope `{scope}` is a special filesystem entry"
        ))),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(DispatchServiceError::InvalidRequest(format!(
            "cannot lstat scope `{scope}`: {error}"
        ))),
    }
}

fn validate_run_artifact(run: &RunId, path: &PathAuthority) -> DispatchServiceResult<()> {
    let prefix = format!(".shepherd/runs/{}/", run.as_str());
    if !path.is_exact() || !path.as_str().starts_with(&prefix) {
        return Err(DispatchServiceError::InvalidRequest(format!(
            "artifact `{path}` must be an exact path under the selected run"
        )));
    }
    Ok(())
}

fn validate_artifact_nofollow(
    root: Option<&Path>,
    artifact: &PathAuthority,
) -> DispatchServiceResult<()> {
    let root = root.ok_or_else(|| {
        DispatchServiceError::InvalidRequest(
            "native artifacts require the bound workspace root".into(),
        )
    })?;
    let path = root.join(artifact.as_str());
    match fs::symlink_metadata(&path) {
        Ok(metadata) if metadata.file_type().is_symlink() => Err(
            DispatchServiceError::InvalidRequest(format!("artifact `{artifact}` is a symlink")),
        ),
        Ok(metadata) if !metadata.is_file() => Err(DispatchServiceError::InvalidRequest(format!(
            "artifact `{artifact}` is not a regular file"
        ))),
        Ok(_) => {
            if !path_is_regular_nofollow(&path)? {
                return Err(DispatchServiceError::InvalidRequest(format!(
                    "artifact `{artifact}` is not a regular file"
                )));
            }
            Ok(())
        }
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(DispatchServiceError::InvalidRequest(format!(
            "cannot inspect artifact `{artifact}`: {error}"
        ))),
    }
}

fn agent_type_matches(target: Harness, role: Role, agent_type: &AgentType) -> bool {
    match target {
        Harness::ClaudeCode => {
            agent_type.as_str() == role.as_str() || agent_type.as_str() == role.carrier()
        }
        Harness::Codex => {
            (role.write_eligible() && agent_type.as_str() == "worker")
                || (!role.write_eligible() && agent_type.as_str() == "explorer")
        }
        Harness::Pi => {
            agent_type.as_str() == format!("pi-subagents:{}", role.as_str())
                || agent_type.as_str() == role.carrier()
        }
        Harness::PrimeAgent => false,
        _ => false,
    }
}

fn validate_installed_attachment(
    loaded: &LoadedCarrierAttestationV1,
    expected: &CarrierAttachmentExpectation,
    installed: &InstalledPackageBinding,
) -> DispatchServiceResult<()> {
    validate_attachment_files(expected, installed)?;
    if loaded.target != expected.target {
        return Err(DispatchServiceError::Domain(
            DispatchError::AttachmentMismatch(
                "loaded carrier target differs from expected target".into(),
            ),
        ));
    }
    Ok(())
}

fn validate_attachment_files(
    expected: &CarrierAttachmentExpectation,
    installed: &InstalledPackageBinding,
) -> DispatchServiceResult<()> {
    // Reopen only the retained package and reproduce its complete compiler
    // inventory. Carrier-parent searches are not authority and miss Codex's
    // separate .agents/skills tree.
    let actual = trusted_attachment_expectation(
        installed,
        CarrierAttachmentExpectationRequest {
            target: expected.target,
            role: expected.role.to_string(),
            agent_id: expected.agent_id.to_string(),
            attachment_kind: attachment_kind_name(expected.attachment_kind).into(),
        },
    )?;
    if !same_attachment(&actual, expected) {
        return Err(DispatchServiceError::Domain(
            DispatchError::AttachmentMismatch(
                "trusted installed attachment changed after prepare".into(),
            ),
        ));
    }
    Ok(())
}

fn harness_name(harness: Harness) -> &'static str {
    match harness {
        Harness::ClaudeCode => "claude",
        Harness::Codex => "codex",
        Harness::Pi => "pi",
        Harness::PrimeAgent => "prime-agent",
        _ => "unknown",
    }
}

fn path_is_regular_nofollow(path: &Path) -> DispatchServiceResult<bool> {
    #[cfg(unix)]
    {
        let descriptor = open_path_nofollow(path)?;
        let stat = rustix::fs::fstat(&descriptor).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot inspect no-follow path {}: {error}",
                path.display()
            ))
        })?;
        Ok(rustix::fs::FileType::from_raw_mode(stat.st_mode).is_file())
    }
    #[cfg(not(unix))]
    {
        reject_symlink_components(path)?;
        let metadata = fs::symlink_metadata(path).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot inspect no-follow path {}: {error}",
                path.display()
            ))
        })?;
        if metadata.file_type().is_symlink() {
            return Err(DispatchServiceError::InvalidRequest(
                "no-follow path is a symlink".into(),
            ));
        }
        Ok(metadata.is_file())
    }
}

fn path_is_directory_nofollow(path: &Path) -> DispatchServiceResult<bool> {
    #[cfg(unix)]
    {
        let descriptor = open_path_nofollow(path)?;
        let stat = rustix::fs::fstat(&descriptor).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot inspect no-follow path {}: {error}",
                path.display()
            ))
        })?;
        Ok(rustix::fs::FileType::from_raw_mode(stat.st_mode).is_dir())
    }
    #[cfg(not(unix))]
    {
        reject_symlink_components(path)?;
        let metadata = fs::symlink_metadata(path).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot inspect no-follow path {}: {error}",
                path.display()
            ))
        })?;
        if metadata.file_type().is_symlink() {
            return Err(DispatchServiceError::InvalidRequest(
                "no-follow path is a symlink".into(),
            ));
        }
        Ok(metadata.is_dir())
    }
}

#[cfg(windows)]
#[allow(unsafe_code)]
fn windows_file_identity(path: &Path) -> DispatchServiceResult<(u32, u64, u32)> {
    use std::os::windows::{fs::OpenOptionsExt, io::AsRawHandle};
    use windows_sys::Win32::Storage::FileSystem::{
        BY_HANDLE_FILE_INFORMATION, FILE_ATTRIBUTE_REPARSE_POINT, FILE_FLAG_BACKUP_SEMANTICS,
        FILE_FLAG_OPEN_REPARSE_POINT, GetFileInformationByHandle,
    };
    let metadata = fs::symlink_metadata(path).map_err(|error| {
        DispatchServiceError::InvalidRequest(format!(
            "cannot inspect Windows file identity {}: {error}",
            path.display()
        ))
    })?;
    let flags = FILE_FLAG_OPEN_REPARSE_POINT
        | if metadata.is_dir() {
            FILE_FLAG_BACKUP_SEMANTICS
        } else {
            0
        };
    let file = fs::OpenOptions::new()
        .read(true)
        .custom_flags(flags)
        .open(path)
        .map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot open Windows file identity {}: {error}",
                path.display()
            ))
        })?;
    let mut information = BY_HANDLE_FILE_INFORMATION::default();
    let result = unsafe { GetFileInformationByHandle(file.as_raw_handle() as _, &mut information) };
    if result == 0 {
        return Err(DispatchServiceError::InvalidRequest(format!(
            "cannot inspect Windows file identity {}",
            path.display()
        )));
    }
    if information.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT != 0 {
        return Err(DispatchServiceError::InvalidRequest(format!(
            "Windows reparse point is not an attachment: {}",
            path.display()
        )));
    }
    let file_id =
        (u64::from(information.nFileIndexHigh) << 32) | u64::from(information.nFileIndexLow);
    Ok((
        information.dwVolumeSerialNumber,
        file_id,
        information.nNumberOfLinks,
    ))
}

/// `Stat::st_mode` is `u16` on macOS and `u32` on Linux, so a plain `u32::from`
/// widens on one target and is a no-op the lint rejects on the other. Routing
/// through `u64`, which is wider than both, states the conversion once and
/// keeps the four-byte mode encoding these digests already commit to.
#[cfg(unix)]
fn mode_bits(mode: impl Into<u64>) -> u32 {
    u32::try_from(mode.into()).unwrap_or(u32::MAX)
}

#[cfg(unix)]
fn metadata_u64<T>(value: T, label: &str) -> DispatchServiceResult<u64>
where
    T: TryInto<u64>,
{
    value.try_into().map_err(|_| {
        DispatchServiceError::InvalidRequest(format!("native {label} metadata overflows u64"))
    })
}

fn hash_carrier_path(path: &Path) -> DispatchServiceResult<[u8; 32]> {
    #[cfg(unix)]
    {
        use rustix::fs::{FileType, fstat};
        let descriptor = open_path_nofollow(path)?;
        let stat = fstat(&descriptor).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot inspect carrier path {}: {error}",
                path.display()
            ))
        })?;
        let kind = FileType::from_raw_mode(stat.st_mode);
        if kind.is_file() && stat.st_nlink != 1 {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "carrier file has multiple hard links: {}",
                path.display()
            )));
        }
        if !kind.is_file() && !kind.is_dir() {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "carrier path is not a regular file or directory: {}",
                path.display()
            )));
        }

        let content_digest: [u8; 32] = if kind.is_file() {
            let bytes = read_bounded_file(File::from(descriptor), MAX_ATTESTATION_BYTES)?;
            let mut content = Sha256::new();
            content.update(b"regular-file\0");
            content.update(mode_bits(stat.st_mode).to_be_bytes());
            content.update(metadata_u64(stat.st_size, "size")?.to_be_bytes());
            content.update(bytes);
            content.finalize().into()
        } else {
            let device = stat.st_dev;
            let mut files = Vec::new();
            let mut total_bytes = 0_usize;
            hash_directory(descriptor, device, "", 0, &mut total_bytes, &mut files)?;
            files.sort_by(|left, right| left.0.cmp(&right.0));
            let mut content = Sha256::new();
            for (relative, mode, bytes) in files {
                content.update(relative.as_bytes());
                content.update([0]);
                content.update(mode.to_be_bytes());
                content.update([0]);
                content.update(bytes);
                content.update([0, 0]);
            }
            content.finalize().into()
        };

        let mut identity = Sha256::new();
        identity.update(b"unix-identity\0");
        identity.update(metadata_u64(stat.st_dev, "device")?.to_be_bytes());
        identity.update(metadata_u64(stat.st_ino, "inode")?.to_be_bytes());
        identity.update(metadata_u64(stat.st_nlink, "link-count")?.to_be_bytes());
        identity.update(mode_bits(stat.st_mode).to_be_bytes());
        identity.update(metadata_u64(stat.st_size, "size")?.to_be_bytes());

        let mut digest = Sha256::new();
        digest.update(b"carrier-identity/1\0");
        digest.update(content_digest);
        digest.update(identity.finalize());
        Ok(digest.finalize().into())
    }
    #[cfg(windows)]
    {
        let metadata = fs::symlink_metadata(path).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot inspect carrier path {}: {error}",
                path.display()
            ))
        })?;
        let content_digest = hash_path(path)?;
        let (volume, file_id, links) = windows_file_identity(path)?;
        if metadata.is_file() && links != 1 {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "carrier file has multiple hard links: {}",
                path.display()
            )));
        }
        if !metadata.is_file() && !metadata.is_dir() {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "carrier path is not a regular file or directory: {}",
                path.display()
            )));
        }
        let mut identity = Sha256::new();
        identity.update(b"windows-identity\0");
        identity.update(u64::from(volume).to_be_bytes());
        identity.update(file_id.to_be_bytes());
        identity.update(u64::from(links).to_be_bytes());
        identity.update(metadata.len().to_be_bytes());
        let mut digest = Sha256::new();
        digest.update(b"carrier-identity/1\0");
        digest.update(content_digest);
        digest.update(identity.finalize());
        Ok(digest.finalize().into())
    }
    #[cfg(not(any(unix, windows)))]
    {
        let _ = path;
        Err(DispatchServiceError::InvalidRequest(
            "native carrier identity is unsupported on this platform".into(),
        ))
    }
}

#[cfg(windows)]
fn hash_path(path: &Path) -> DispatchServiceResult<[u8; 32]> {
    hash_path_with_prefix(path, "")
}

fn hash_skill_bundle(path: &Path, skill: &str) -> DispatchServiceResult<[u8; 32]> {
    if !path_is_directory_nofollow(path)? {
        return Err(attachment_mismatch(
            "installed skill bundle must be a directory",
        ));
    }
    // The compiler normalizes every harness bundle to skills/<name>/ before
    // hashing its complete file inventory. Preserve that same namespace here.
    hash_path_with_prefix(path, &format!("skills/{skill}/"))
}

fn hash_path_with_prefix(path: &Path, prefix: &str) -> DispatchServiceResult<[u8; 32]> {
    #[cfg(unix)]
    {
        use rustix::fs::{FileType, fstat};
        let descriptor = open_path_nofollow(path)?;
        let stat = fstat(&descriptor).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot inspect carrier path {}: {error}",
                path.display()
            ))
        })?;
        if FileType::from_raw_mode(stat.st_mode).is_file() {
            if stat.st_nlink != 1 {
                return Err(DispatchServiceError::InvalidRequest(format!(
                    "carrier file has multiple hard links: {}",
                    path.display()
                )));
            }
            let bytes = read_bounded_file(File::from(descriptor), MAX_ATTESTATION_BYTES)?;
            let mut digest = Sha256::new();
            digest.update(b"regular-file\0");
            digest.update(mode_bits(stat.st_mode).to_be_bytes());
            digest.update(stat.st_size.to_be_bytes());
            digest.update(&bytes);
            return Ok(digest.finalize().into());
        }
        if !FileType::from_raw_mode(stat.st_mode).is_dir() {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "carrier path is not a regular file or directory: {}",
                path.display()
            )));
        }
        let device = stat.st_dev;
        let mut files = Vec::new();
        let mut total_bytes = 0_usize;
        hash_directory(descriptor, device, "", 0, &mut total_bytes, &mut files)?;
        files.sort_by(|left, right| left.0.cmp(&right.0));
        let mut digest = Sha256::new();
        for (relative, mode, bytes) in files {
            digest.update(prefix.as_bytes());
            digest.update(relative.as_bytes());
            digest.update([0]);
            digest.update(mode.to_be_bytes());
            digest.update([0]);
            digest.update(bytes);
            digest.update([0, 0]);
        }
        Ok(digest.finalize().into())
    }
    #[cfg(not(unix))]
    {
        reject_symlink_components(path)?;
        let metadata = fs::symlink_metadata(path).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot inspect carrier path {}: {error}",
                path.display()
            ))
        })?;
        if metadata.file_type().is_symlink() {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "carrier path is a symlink: {}",
                path.display()
            )));
        }
        if metadata.is_file() {
            let (_, _, links) = windows_file_identity(path)?;
            if links != 1 {
                return Err(DispatchServiceError::InvalidRequest(format!(
                    "carrier file has multiple hard links: {}",
                    path.display()
                )));
            }
            let bytes = read_path_nofollow(path, MAX_ATTESTATION_BYTES)?;
            let mut digest = Sha256::new();
            digest.update(b"regular-file\0");
            digest.update(0o644_u32.to_be_bytes());
            digest.update(metadata.len().to_be_bytes());
            digest.update(&bytes);
            return Ok(digest.finalize().into());
        }
        if !metadata.is_dir() {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "carrier path is not a regular file or directory: {}",
                path.display()
            )));
        }
        let mut files = Vec::new();
        let mut total_bytes = 0_usize;
        collect_tree(path, Path::new(""), 0, &mut total_bytes, &mut files)?;
        files.sort_by(|left, right| left.0.cmp(&right.0));
        let mut digest = Sha256::new();
        for (relative, mode, bytes) in files {
            digest.update(prefix.as_bytes());
            digest.update(relative.as_bytes());
            digest.update([0]);
            digest.update(mode.to_be_bytes());
            digest.update([0]);
            digest.update(bytes);
            digest.update([0, 0]);
        }
        Ok(digest.finalize().into())
    }
}

#[cfg(unix)]
fn hash_directory(
    descriptor: rustix::fd::OwnedFd,
    device: rustix::fs::Dev,
    prefix: &str,
    depth: usize,
    total_bytes: &mut usize,
    files: &mut Vec<(String, u32, Vec<u8>)>,
) -> DispatchServiceResult<()> {
    use rustix::fs::{Dir, OFlags, openat};
    if depth > MAX_ATTACHMENT_DEPTH {
        return Err(DispatchServiceError::InvalidRequest(
            "carrier tree exceeds the native depth limit".into(),
        ));
    }
    let mut directory = Dir::new(descriptor).map_err(|error| {
        DispatchServiceError::InvalidRequest(format!("cannot open carrier directory: {error}"))
    })?;
    let mut names = Vec::new();
    for entry in &mut directory {
        let entry = entry.map_err(|error| {
            DispatchServiceError::InvalidRequest(format!("cannot enumerate carrier tree: {error}"))
        })?;
        let name = entry.file_name().to_str().map_err(|_| {
            DispatchServiceError::InvalidRequest(
                "carrier tree contains a non-UTF-8 file name".into(),
            )
        })?;
        if matches!(name, "." | "..") {
            continue;
        }
        names.push(name.to_owned());
    }
    names.sort();
    for name in names {
        let child = if prefix.is_empty() {
            name.clone()
        } else {
            format!("{prefix}/{name}")
        };
        let fd = openat(
            directory.fd().map_err(|error| {
                DispatchServiceError::InvalidRequest(format!(
                    "cannot retain carrier directory: {error}"
                ))
            })?,
            &name,
            OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW | OFlags::NONBLOCK,
            rustix::fs::Mode::empty(),
        )
        .map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot open carrier entry `{child}` without following links: {error}"
            ))
        })?;
        let stat = rustix::fs::fstat(&fd).map_err(|error| {
            DispatchServiceError::InvalidRequest(format!(
                "cannot stat carrier entry `{child}`: {error}"
            ))
        })?;
        let kind = rustix::fs::FileType::from_raw_mode(stat.st_mode);
        if stat.st_dev != device {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "carrier tree crosses a filesystem mount at `{child}`"
            )));
        }
        if kind.is_dir() {
            hash_directory(fd, device, &child, depth + 1, total_bytes, files)?;
        } else if kind.is_file() {
            if stat.st_nlink != 1 {
                return Err(DispatchServiceError::InvalidRequest(format!(
                    "carrier tree file `{child}` has multiple hard links"
                )));
            }
            if files.len() >= MAX_ATTACHMENT_FILES {
                return Err(DispatchServiceError::InvalidRequest(
                    "carrier tree exceeds the native file-count limit".into(),
                ));
            }
            let mode = mode_bits(stat.st_mode) & 0o777;
            let bytes = read_bounded_file(File::from(fd), MAX_ATTESTATION_BYTES)?;
            *total_bytes = total_bytes.checked_add(bytes.len()).ok_or_else(|| {
                DispatchServiceError::InvalidRequest("carrier tree byte count overflow".into())
            })?;
            if *total_bytes > MAX_ATTACHMENT_TREE_BYTES {
                return Err(DispatchServiceError::InvalidRequest(
                    "carrier tree exceeds the native byte limit".into(),
                ));
            }
            files.push((child, mode, bytes));
        } else {
            return Err(DispatchServiceError::InvalidRequest(format!(
                "carrier tree contains unsupported entry `{child}`"
            )));
        }
    }
    Ok(())
}

#[cfg(not(unix))]
fn collect_tree(
    root: &Path,
    relative: &Path,
    depth: usize,
    total_bytes: &mut usize,
    files: &mut Vec<(String, u32, Vec<u8>)>,
) -> DispatchServiceResult<()> {
    if depth > MAX_ATTACHMENT_DEPTH {
        return Err(DispatchServiceError::InvalidRequest(
            "carrier tree exceeds the native depth limit".into(),
        ));
    }
    if files.len() >= MAX_ATTACHMENT_FILES {
        return Err(DispatchServiceError::InvalidRequest(
            "carrier tree exceeds the native file-count limit".into(),
        ));
    }
    let current = root.join(relative);
    let mut entries = fs::read_dir(&current)
        .map_err(|error| DispatchServiceError::InvalidRequest(error.to_string()))?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|error| DispatchServiceError::InvalidRequest(error.to_string()))?;
    entries.sort_by_key(|entry| entry.file_name());
    for entry in entries {
        let path = entry.path();
        let metadata = fs::symlink_metadata(&path)
            .map_err(|error| DispatchServiceError::InvalidRequest(error.to_string()))?;
        if metadata.file_type().is_symlink() {
            return Err(DispatchServiceError::InvalidRequest(
                "carrier tree contains a symlink".into(),
            ));
        }
        let child = relative.join(entry.file_name());
        if metadata.is_dir() {
            collect_tree(root, &child, depth + 1, total_bytes, files)?;
        } else if metadata.is_file() {
            let (_, _, links) = windows_file_identity(&path)?;
            if metadata.file_type().is_symlink() || links != 1 {
                return Err(DispatchServiceError::InvalidRequest(
                    "carrier tree file has unsafe link identity".into(),
                ));
            }
            let bytes = read_path_nofollow(&path, MAX_ATTESTATION_BYTES)?;
            *total_bytes = total_bytes.checked_add(bytes.len()).ok_or_else(|| {
                DispatchServiceError::InvalidRequest("carrier tree byte count overflow".into())
            })?;
            if *total_bytes > MAX_ATTACHMENT_TREE_BYTES {
                return Err(DispatchServiceError::InvalidRequest(
                    "carrier tree exceeds the native byte limit".into(),
                ));
            }
            let mode = 0o644;
            files.push((child.to_string_lossy().replace('\\', "/"), mode, bytes));
        } else {
            return Err(DispatchServiceError::InvalidRequest(
                "carrier tree contains unsupported entry".into(),
            ));
        }
    }
    Ok(())
}

fn same_root_identity(existing: &RootSessionBinding, requested: &RootSessionBinding) -> bool {
    same_root_principal(existing, requested) && existing.mode == requested.mode
}

fn same_root_principal(existing: &RootSessionBinding, requested: &RootSessionBinding) -> bool {
    existing.schema == requested.schema
        && existing.project_id == requested.project_id
        && existing.run == requested.run
        && existing.harness == requested.harness
        && existing.session_id == requested.session_id
        && existing.role == requested.role
        && existing.project_filesystem_id == requested.project_filesystem_id
}

fn can_transition_root_to_execution(
    existing: &RootSessionBinding,
    requested: &RootSessionBinding,
) -> bool {
    same_root_principal(existing, requested)
        && existing.mode == "planning"
        && requested.mode == "execution"
        && requested.bound_at > existing.bound_at
}

fn lease_expires_at(lease_ms: u64, now: i64) -> DispatchServiceResult<i64> {
    if lease_ms == 0 || lease_ms > MAX_LEASE_MS {
        return Err(DispatchServiceError::InvalidRequest(format!(
            "lease_ms must be between 1 and {MAX_LEASE_MS}"
        )));
    }
    let lease_ms = i64::try_from(lease_ms)
        .map_err(|_| DispatchServiceError::InvalidRequest("lease_ms overflow".into()))?;
    now.checked_add(lease_ms)
        .ok_or_else(|| DispatchServiceError::InvalidRequest("lease time overflow".into()))
}

fn validate_schema(schema: &str) -> DispatchServiceResult<()> {
    if schema == REQUEST_SCHEMA {
        Ok(())
    } else {
        Err(DispatchServiceError::InvalidRequest(format!(
            "unsupported schema `{schema}`"
        )))
    }
}

fn validate_requested_binding_run(
    requested: Option<&str>,
    bound: &RunId,
) -> DispatchServiceResult<()> {
    let Some(requested) = requested else {
        return Ok(());
    };
    let requested = RunId::new(requested)?;
    if &requested == bound {
        Ok(())
    } else {
        Err(DispatchServiceError::InvalidRequest(format!(
            "requested root run `{requested}` does not match newest explicit binding `{bound}`"
        )))
    }
}

#[cfg(test)]
mod security_tests {

    /// Remove a fixture tree, tolerating Windows' delayed handle release.
    ///
    /// A directory cannot be removed on Windows while any file inside it still
    /// has an open handle, and a handle closed by an exiting child is not always
    /// released by the time the parent's next syscall runs. These tests failed
    /// their teardown rather than their assertions with `os error 32`. Bounding
    /// the retry is what keeps it from hiding a genuine handle leak.
    ///
    /// Mirrors `crates/cli/tests/support/mod.rs`, which cannot be imported here
    /// because every `tests/*.rs` is its own crate.
    fn remove_fixture(path: impl AsRef<std::path::Path>) {
        const ATTEMPTS: u32 = 100;

        let path = path.as_ref();
        let mut last = None;
        for attempt in 0..ATTEMPTS {
            match std::fs::remove_dir_all(path) {
                Ok(()) => return,
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => return,
                Err(error) => {
                    last = Some(error);
                    if attempt + 1 < ATTEMPTS {
                        std::thread::sleep(std::time::Duration::from_millis(100));
                    }
                }
            }
        }
        panic!(
            "cannot remove fixture {} after {ATTEMPTS} attempts: {}",
            path.display(),
            last.expect("a failure was recorded")
        );
    }
    use super::*;
    #[cfg(unix)]
    use std::os::unix::ffi::OsStrExt;

    fn root_owned_record(run: &str, root_session: &str) -> DispatchRecord {
        root_owned_role_record(run, root_session, Role::Worker)
    }

    fn root_owned_role_record(run: &str, root_session: &str, role: Role) -> DispatchRecord {
        let contract = role
            .dispatch_capability_contract()
            .expect("role capability contract");
        let observed = contract
            .required
            .union(&contract.optional)
            .cloned()
            .collect::<BTreeSet<_>>();
        DispatchRecord::start(DispatchStart {
            project_id: ProjectId::new("018f47ce-72d7-7f64-9eb1-2f651d521c2a").expect("project id"),
            run: RunId::new(run).expect("run id"),
            root_session_id: SessionId::new(root_session).expect("root session id"),
            run_incarnation: format!("incarnation-{run}"),
            nonce: format!("nonce-{run}"),
            harness: Harness::Codex,
            agent_id: AgentId::new(format!("{role}-{run}")).expect("agent id"),
            agent_type: AgentType::new("worker").expect("agent type"),
            role,
            lane: (role != Role::Engineer).then(|| LaneId::new("lane-a").expect("lane")),
            parent_agent_id: None,
            session_id: SessionId::new(format!("worker-session-{run}")).expect("worker session"),
            write_scope: if role == Role::Engineer {
                vec![format!(".shepherd/runs/{run}/plan.md")]
            } else {
                vec!["docs/result.md".into()]
            },
            model: None,
            capability_contract: contract,
            capability_probe: CapabilityProbe::new(
                observed,
                "native-root-revocation",
                "native",
                None,
                100,
            )
            .expect("capability probe"),
            startup_attachment: None,
            attachment_nonce: None,
            result_artifact: None,
            result_nonce: None,
            review_artifact: None,
            review_nonce: None,
            started_at: 100,
            lease_expires_at: 10_000,
            resumes_agent_id: None,
        })
        .expect("dispatch record")
    }

    fn review_pending(
        agent: &str,
        session: &str,
        launch_marker: u8,
        replaces: Option<&str>,
        result_artifact: &str,
    ) -> PendingDispatch {
        PendingDispatch {
            schema: shepherd::dispatch::PENDING_DISPATCH_SCHEMA.into(),
            launch_id_hash: [launch_marker; 32],
            parent_process_hash: [2; 32],
            project_id: ProjectId::new("018f47ce-72d7-7f64-9eb1-2f651d521c2a").expect("project"),
            project_filesystem_id: ProjectFilesystemId::new("03".repeat(32)).expect("filesystem"),
            run: RunId::new("v645").expect("run"),
            run_status: "executing".into(),
            root_session_id: SessionId::new("review-root").expect("root"),
            caller_role: Role::Shepherd,
            parent_dispatch_id: None,
            replaces_agent_id: replaces.map(AgentId::new).transpose().expect("lineage"),
            role: Role::Conductor,
            work_kind: WorkKind::Coordination,
            lane: Some(LaneId::new("lane-a").expect("lane")),
            baseline_commit: GitCommit::new("04".repeat(20)).expect("commit"),
            read_scope: vec![PathAuthority::new("docs/**").expect("read")],
            write_scope: vec![],
            result_artifact: PathAuthority::exact(result_artifact).expect("result"),
            review_artifact: PathAuthority::exact(
                ".shepherd/runs/v645/lanes/lane-a/reviews/conductor.json",
            )
            .expect("review"),
            task_path: PathAuthority::exact("docs/task.md").expect("task"),
            task_sha256: [5; 32],
            expected_child_session_id: SessionId::new(session).expect("child session"),
            expected_attachment: CarrierAttachmentExpectation {
                target: Harness::Codex,
                role: Role::Conductor,
                agent_id: AgentId::new(agent).expect("agent"),
                installed_carrier_path: "/private/tmp/conductor.md".into(),
                candidate_sha256: [9; 32],
                carrier_sha256: [6; 32],
                compiler_tree_sha256: [7; 32],
                startup_skill: "coordination".into(),
                skill_bundle_sha256: [8; 32],
                attachment_kind: AttachmentKind::CodexCustomAgent,
            },
            expires_at: 10_000,
            launch_state: PendingLaunchState::Pending,
            claimed_at: None,
            child_process_hash: None,
            activated_at: None,
            nonce_sha256: [9; 32],
        }
    }

    fn active_review_record(pending: &PendingDispatch) -> DispatchRecord {
        let contract = pending
            .role
            .dispatch_capability_contract()
            .expect("capability contract");
        let observed = contract
            .required
            .union(&contract.optional)
            .cloned()
            .collect::<BTreeSet<_>>();
        DispatchRecord::start(DispatchStart {
            project_id: pending.project_id.clone(),
            run: pending.run.clone(),
            root_session_id: pending.root_session_id.clone(),
            run_incarnation: "incarnation-v645".into(),
            nonce: format!("nonce-{}", pending.expected_attachment.agent_id),
            harness: pending.expected_attachment.target,
            agent_id: pending.expected_attachment.agent_id.clone(),
            agent_type: AgentType::new(pending.role.as_str()).expect("agent type"),
            role: pending.role,
            lane: pending.lane.clone(),
            parent_agent_id: pending
                .parent_dispatch_id
                .as_ref()
                .map(|id| AgentId::new(id.as_str()).expect("parent agent")),
            session_id: pending.expected_child_session_id.clone(),
            write_scope: pending
                .write_scope
                .iter()
                .map(ToString::to_string)
                .collect(),
            model: None,
            capability_contract: contract,
            capability_probe: CapabilityProbe::new(observed, "fixture", "fixture", None, 101)
                .expect("probe"),
            startup_attachment: None,
            attachment_nonce: None,
            result_artifact: Some(pending.result_artifact.as_str().into()),
            result_nonce: Some("ab".repeat(32)),
            review_artifact: None,
            review_nonce: None,
            started_at: 101,
            lease_expires_at: pending.expires_at,
            resumes_agent_id: None,
        })
        .expect("subject record")
    }

    fn auditor_record() -> DispatchRecord {
        let role = Role::Auditor;
        let contract = role
            .dispatch_capability_contract()
            .expect("Auditor contract");
        let observed = contract
            .required
            .union(&contract.optional)
            .cloned()
            .collect::<BTreeSet<_>>();
        DispatchRecord::start(DispatchStart {
            project_id: ProjectId::new("018f47ce-72d7-7f64-9eb1-2f651d521c2a").expect("project"),
            run: RunId::new("v645").expect("run"),
            root_session_id: SessionId::new("review-root").expect("root"),
            run_incarnation: "incarnation-v645".into(),
            nonce: "auditor-nonce".into(),
            harness: Harness::Codex,
            agent_id: AgentId::new("auditor-agent").expect("agent"),
            agent_type: AgentType::new("auditor").expect("agent type"),
            role,
            lane: Some(LaneId::new("lane-a").expect("lane")),
            parent_agent_id: Some(AgentId::new("subject-agent").expect("parent")),
            session_id: SessionId::new("auditor-session").expect("session"),
            write_scope: vec![".shepherd/runs/v645/lanes/lane-a/reviews/auditor.json".into()],
            model: None,
            capability_contract: contract,
            capability_probe: CapabilityProbe::new(observed, "fixture", "fixture", None, 102)
                .expect("probe"),
            startup_attachment: None,
            attachment_nonce: None,
            result_artifact: Some(".shepherd/runs/v645/lanes/lane-a/reports/auditor.json".into()),
            result_nonce: Some("bc".repeat(32)),
            review_artifact: Some(".shepherd/runs/v645/lanes/lane-a/reviews/auditor.json".into()),
            review_nonce: Some("cd".repeat(32)),
            started_at: 102,
            lease_expires_at: 10_000,
            resumes_agent_id: None,
        })
        .expect("auditor record")
    }

    fn redo_review(marker: u8) -> ReviewResult {
        ReviewResult {
            schema: shepherd::dispatch::REVIEW_RESULT_SCHEMA.into(),
            run: RunId::new("v645").expect("run"),
            lane: Some(LaneId::new("lane-a").expect("lane")),
            mode: shepherd::dispatch::ReviewMode::AuditorPosthoc,
            reviewer_role: Role::Auditor,
            candidate_commit: "0123456789abcdef0123456789abcdef01234567".into(),
            input_digest: "aa".repeat(32),
            startup_skill: "reviewing".into(),
            skill_bundle_digest: "bb".repeat(32),
            result_channel: "native-result".into(),
            verdict: shepherd::dispatch::ReviewVerdict::Redo,
            findings: vec![shepherd::dispatch::ReviewFinding {
                finding_id: format!("finding-{marker}"),
                location: "docs/task.md:1".into(),
                hypothesis: "owned output fails its acceptance predicate".into(),
                falsification_command: "shepherd verify".into(),
                falsification_exit_status: 1,
                observed_result: format!("failure-{marker}"),
                confidence: shepherd::dispatch::ReviewConfidence::StructurallyVerifiable,
                severity: shepherd::dispatch::ReviewSeverity::Important,
                impact: "owned output remains red".into(),
                acceptance_predicate: "focused verification exits zero".into(),
                owner_role: Role::Conductor,
                route: "redo subject".into(),
                evidence_paths: vec![format!("evidence/failure-{marker}.txt")],
            }],
            report_path: Some(".shepherd/runs/v645/lanes/lane-a/reviews/auditor.json".into()),
        }
    }

    fn planning_review_pending(
        agent: &str,
        marker: u8,
        critic: bool,
        replaces: Option<&str>,
    ) -> PendingDispatch {
        let role = if critic { Role::Critic } else { Role::Engineer };
        let result = if critic {
            ".shepherd/runs/v645/reports/planning-critic.json"
        } else {
            ".shepherd/runs/v645/phase0.md"
        };
        let mut pending =
            review_pending(agent, &format!("{agent}-session"), marker, replaces, result);
        pending.run_status = "planted".into();
        pending.role = role;
        pending.work_kind = if critic {
            WorkKind::Review
        } else {
            WorkKind::Planning
        };
        pending.lane = None;
        pending.expected_attachment.role = role;
        pending.expected_attachment.installed_carrier_path = format!("/private/tmp/{role}.md");
        pending.expected_attachment.startup_skill =
            if critic { "reviewing" } else { "planning" }.into();
        pending.review_artifact = PathAuthority::exact(format!(
            ".shepherd/runs/v645/reports/planning-{role}-review.json"
        ))
        .expect("review path");
        if critic {
            pending.caller_role = Role::Engineer;
            pending.parent_dispatch_id =
                Some(shepherd::dispatch::DispatchId::new("planning-engineer").expect("parent"));
        } else {
            pending.write_scope = vec![PathAuthority::exact(result).expect("planning scope")];
        }
        pending.validate().expect("typed planning pending fixture");
        pending
    }

    // These isolated files are test inputs, never live carrier proof. Every
    // ruling, quarantine, and replacement below uses the production service.
    fn planning_review_fixture(
        mode: &str,
        status: &str,
    ) -> (PathBuf, DispatchService, PendingDispatch, PendingDispatch) {
        static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("clock")
            .as_nanos();
        let ordinal = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let fixture = std::env::temp_dir().join(format!(
            "shepherd-planning-review-{}-{nonce:x}-{ordinal}",
            std::process::id()
        ));
        std::fs::create_dir(&fixture).expect("new isolated fixture");
        let fixture = std::fs::canonicalize(fixture).expect("canonical fixture");
        let runs = fixture.join("runs");
        let mut state: shepherd::RunState = serde_json::from_value(serde_json::json!({
            "run": "v645", "status": "planted",
        }))
        .expect("run fixture");
        state
            .store(&runs.join("v645/run.json"))
            .expect("persist run fixture");
        let project = ProjectId::new("018f47ce-72d7-7f64-9eb1-2f651d521c2a").expect("project");
        let store = DispatchStore::new(&runs);
        let binding = RootSessionBinding {
            schema: ROOT_SESSION_SCHEMA.into(),
            project_id: project.clone(),
            run: RunId::new("v645").expect("run"),
            harness: Harness::Codex,
            session_id: SessionId::new("review-root").expect("root"),
            role: Role::Shepherd,
            mode: mode.into(),
            project_filesystem_id: None,
            bound_at: 1,
            expires_at: 10_000,
        };
        store
            .publish_root_binding_for_run(&binding)
            .expect("root fixture");
        store
            .activate_current_root_binding(&binding)
            .expect("current root fixture");
        let mut subject = planning_review_pending("planning-engineer", 31, false, None);
        let mut reviewer = planning_review_pending("planning-critic", 32, true, None);
        for pending in [&mut subject, &mut reviewer] {
            pending.claim(100, [10; 32]).expect("claim fixture");
            pending.activate(101, [10; 32]).expect("activate fixture");
            store.publish_pending(pending).expect("pending fixture");
            std::fs::write(
                runs.join(format!(
                    "v645/dispatch/{}.json",
                    pending.expected_attachment.agent_id
                )),
                serde_json::to_vec(&active_review_record(pending)).expect("dispatch JSON"),
            )
            .expect("dispatch fixture");
        }
        state.status = status.into();
        state
            .store(&runs.join("v645/run.json"))
            .expect("selected phase fixture");
        (
            fixture,
            DispatchService::new(store, project),
            subject,
            reviewer,
        )
    }

    fn planning_review_ruling(marker: u8) -> ReviewRulingRequest {
        let mut review = redo_review(marker);
        review.lane = None;
        review.mode = shepherd::dispatch::ReviewMode::CriticPrehoc;
        review.reviewer_role = Role::Critic;
        review.verdict = shepherd::dispatch::ReviewVerdict::Red;
        review.report_path = None;
        review.findings[0].owner_role = Role::Engineer;
        ReviewRulingRequest {
            schema: REVIEW_RULING_REQUEST_SCHEMA.into(),
            run: "v645".into(),
            harness: Harness::Codex,
            root_session_id: "review-root".into(),
            subject_agent_id: "planning-engineer".into(),
            reviewer_dispatch_id: "planning-critic".into(),
            reviewer_session_id: "planning-critic-session".into(),
            task_generation: 1,
            review,
        }
    }

    fn published_planning_review_fixture() -> (
        PathBuf,
        DispatchService,
        PendingDispatch,
        DispatchSingletonPublication,
    ) {
        let (fixture, mut service, subject, _) = planning_review_fixture("planning", "planted");
        let db = fixture.join("registry.db");
        Registry::open_migrated(&db)
            .expect("fixture registry")
            .execute(
                "INSERT INTO projects (id, name, created_at, updated_at) VALUES (?1, ?2, ?3, ?3)",
                (
                    service.project_id.as_str(),
                    "planning review fixture",
                    1_i64,
                ),
            )
            .expect("registered fixture project");
        service.registry_path = Some(db);
        let record = service
            .store
            .load_for_run(&subject.run, &subject.expected_attachment.agent_id)
            .expect("active Engineer fixture");
        let (mut registry, publication) = service
            .prepare_singleton_publication(&record, 101, None)
            .expect("prepare singleton")
            .expect("Engineer singleton");
        std::fs::write(
            fixture.join("runs/v645/dispatch/planning-engineer.json"),
            &publication.record_json,
        )
        .expect("canonical fixture publication bytes");
        registry
            .transaction_immediate::<_, RegistryError, _>(|tx| {
                tx.mark_dispatch_singleton_published(&publication.nonce, 101)
            })
            .expect("published singleton");
        let publication = registry
            .load_dispatch_publication(&publication.nonce)
            .expect("publication")
            .expect("published");
        (fixture, service, subject, publication)
    }

    #[test]
    fn planning_review_terminal_publication_preserves_exact_claim_and_history() {
        let (fixture, service, subject, original) = published_planning_review_fixture();
        for round in 1..=4 {
            service
                .review_ruling(planning_review_ruling(round), 200 + i64::from(round))
                .expect("native review ruling");
        }
        let mut registry =
            Registry::open_migrated(service.registry_path.as_ref().unwrap()).expect("registry");
        let publication = registry
            .load_dispatch_publication(&original.nonce)
            .expect("load")
            .expect("history");
        assert_eq!(
            publication.state,
            shepherd::registry::SingletonPublicationState::Published
        );
        let bytes = std::fs::read(fixture.join("runs/v645/dispatch/planning-engineer.json"))
            .expect("terminal bytes");
        assert_eq!(
            publication.record_json.as_bytes(),
            bytes,
            "fourth rejection must refresh the exact publication receipt"
        );
        assert_eq!(publication.record_sha256, hex(&sha256(&bytes)));
        let claim = registry
            .load_dispatch_singleton(service.project_id.as_str(), "v645", "engineer", "__run__")
            .expect("claim")
            .expect("malignant Engineer retains singleton until authorized activation");
        assert_eq!(
            claim.agent_id,
            subject.expected_attachment.agent_id.as_str()
        );
        assert_eq!(
            claim.publication_nonce.as_deref(),
            Some(original.nonce.as_str())
        );
        let snapshot = service
            .store
            .read_review_terminal_snapshot(&subject.run, &subject.expected_attachment.agent_id)
            .unwrap();
        registry.transaction_immediate::<_, RegistryError, _>(|tx| {
            tx.refresh_review_terminal_singleton(&original, &publication.record_json, &snapshot.pending, &snapshot.custody)
        }).expect("stale reader observing the same already-refreshed exact terminal delta is idempotent");
        let report = service
            .store
            .reconcile_singletons(&mut registry, 205)
            .expect("reconcile exact terminal bytes");
        assert_eq!(report.quarantined, 0);
        assert_eq!(report.unchanged, 1);
        assert!(
            service
                .ensure_singleton_is_unclaimed(&subject.run, Role::Engineer, None, None, 205)
                .is_err(),
            "rejection four must not admit an unlineaged Engineer"
        );
        // Close the store's handles first. Windows refuses to remove a
        // directory whose files are still open, and the bounded retry in
        // remove_fixture exists to surface exactly this leak rather than
        // hide it. Unix unlinks open files, so only Windows failed.
        drop(service);
        drop(registry);
        remove_fixture(&fixture);
    }

    #[test]
    fn planning_review_terminal_publication_recovers_only_exact_native_quarantine() {
        let (fixture, service, subject, original) = published_planning_review_fixture();
        // Reproduce a crash after Native's terminal filesystem transaction but
        // before its separate SQLite refresh. No live evidence is fabricated.
        let mut interrupted = service.clone();
        interrupted.registry_path = None;
        for round in 1..=4 {
            interrupted
                .review_ruling(planning_review_ruling(round), 200 + i64::from(round))
                .expect("native terminal transition");
        }
        let mut registry =
            Registry::open_migrated(service.registry_path.as_ref().unwrap()).expect("registry");
        let report = service
            .store
            .reconcile_singletons(&mut registry, 205)
            .expect("recover interrupted publication");
        assert_eq!(
            report.quarantined, 0,
            "exact Native terminal transition is not publication corruption"
        );
        let publication = registry
            .load_dispatch_publication(&original.nonce)
            .expect("load")
            .expect("history retained");
        let bytes = std::fs::read(fixture.join("runs/v645/dispatch/planning-engineer.json"))
            .expect("terminal retained");
        assert_eq!(publication.record_json.as_bytes(), bytes);
        assert_eq!(publication.record_sha256, hex(&sha256(&bytes)));
        assert_eq!(
            registry
                .load_dispatch_singleton(service.project_id.as_str(), "v645", "engineer", "__run__")
                .expect("claim")
                .expect("claim retained")
                .agent_id,
            subject.expected_attachment.agent_id.as_str()
        );
        // Close the store's handles first. Windows refuses to remove a
        // directory whose files are still open, and the bounded retry in
        // remove_fixture exists to surface exactly this leak rather than
        // hide it. Unix unlinks open files, so only Windows failed.
        drop(service);
        drop(registry);
        remove_fixture(&fixture);
    }

    #[test]
    fn planning_review_singleton_transfer_is_activation_atomic_and_replay_safe() {
        for finish_activation in [false, true] {
            let (fixture, service, subject, old_publication) = published_planning_review_fixture();
            for round in 1..=4 {
                service
                    .review_ruling(planning_review_ruling(round), 200 + i64::from(round))
                    .expect("native ruling");
            }
            let mut replacement = planning_review_pending(
                "replacement-engineer",
                35,
                false,
                Some("planning-engineer"),
            );
            service
                .store
                .publish_pending(&replacement)
                .expect("prepared replacement fixture");
            let mut record = active_review_record(&replacement);
            record.started_at = 208;
            record.startup_attachment = Some(StartupAttachment {
                skill: replacement.expected_attachment.startup_skill.clone(),
                bundle_digest: hex(&replacement.expected_attachment.skill_bundle_sha256),
            });
            record.attachment_nonce = Some(hex(&replacement.nonce_sha256));
            record
                .validate_loaded()
                .expect("replacement record fixture");
            let source = service
                .store
                .read_review_terminal_snapshot(&subject.run, &subject.expected_attachment.agent_id)
                .expect("malignant source snapshot");
            assert!(
                service
                    .prepare_singleton_publication(&record, 208, Some(&source))
                    .is_err(),
                "a malignant source is not itself replacement authorization"
            );
            service
                .ensure_singleton_is_unclaimed(
                    &subject.run,
                    Role::Engineer,
                    None,
                    Some((
                        &subject.expected_attachment.agent_id,
                        &subject.root_session_id,
                    )),
                    205,
                )
                .expect("root may prepare exact malignant replacement without releasing the claim");
            service
                .review_replace(
                    ReviewReplacementRequest {
                        schema: REVIEW_REPLACEMENT_REQUEST_SCHEMA.into(),
                        run: "v645".into(),
                        harness: Harness::Codex,
                        root_session_id: "review-root".into(),
                        subject_agent_id: "planning-engineer".into(),
                        replacement_agent_id: "replacement-engineer".into(),
                    },
                    206,
                )
                .expect("native root authorization");
            let source = service
                .store
                .read_review_terminal_snapshot(&subject.run, &subject.expected_attachment.agent_id)
                .expect("authorized source snapshot");
            let mut registry =
                Registry::open_migrated(service.registry_path.as_ref().unwrap()).expect("registry");
            let old = registry
                .load_dispatch_publication(&old_publication.nonce)
                .expect("old")
                .expect("old published");
            for changed in [
                {
                    let mut r = record.clone();
                    r.session_id = source.subject.session_id.clone();
                    r
                },
                {
                    let mut r = record.clone();
                    r.root_session_id = SessionId::new("foreign-root").unwrap();
                    r
                },
                {
                    let mut r = record.clone();
                    r.write_scope.push(".shepherd/runs/v645/plan.md".into());
                    r
                },
                {
                    let mut r = record.clone();
                    r.started_at = 205;
                    r
                },
            ] {
                assert!(
                    service
                        .prepare_singleton_publication(&changed, changed.started_at, Some(&source))
                        .is_err(),
                    "replacement identity, scope, and authorization time are exact"
                );
            }
            let mut forged_source = source.clone();
            forged_source.custody.replacement_agent_id =
                Some(AgentId::new("other-engineer").unwrap());
            assert!(
                service
                    .prepare_singleton_publication(&record, 208, Some(&forged_source))
                    .is_err()
            );
            // A valid JSON/hash pair is insufficient if its bytes differ from
            // the actual Native terminal record. Preserve that old receipt.
            registry.execute("UPDATE dispatch_singleton_publications SET record_json = ?1, record_sha256 = ?2 WHERE nonce = ?3",
                (&old_publication.record_json, &old_publication.record_sha256, &old.nonce)).expect("stale payload fixture");
            assert!(
                service
                    .prepare_singleton_publication(&record, 208, Some(&source))
                    .is_err(),
                "stale source payload denied"
            );
            registry.execute("UPDATE dispatch_singleton_publications SET record_json = ?1, record_sha256 = ?2 WHERE nonce = ?3",
                (&old.record_json, &old.record_sha256, &old.nonce)).expect("restore fixture receipt");
            registry.execute("UPDATE dispatch_singleton_claims SET session_id = 'foreign-session' WHERE agent_id = 'planning-engineer'", ())
                .expect("claim drift fixture");
            assert!(
                service
                    .prepare_singleton_publication(&record, 208, Some(&source))
                    .is_err(),
                "claim metadata drift denied"
            );
            registry.execute("UPDATE dispatch_singleton_claims SET session_id = ?1 WHERE agent_id = 'planning-engineer'",
                [&source.subject.session_id.as_str()]).expect("restore claim fixture");
            let (_, prepared) = service
                .prepare_singleton_publication(&record, 208, Some(&source))
                .expect("authorized intent")
                .expect("singleton");
            let claim = || {
                Registry::open_migrated(service.registry_path.as_ref().unwrap())
                    .expect("registry")
                    .load_dispatch_singleton(
                        service.project_id.as_str(),
                        "v645",
                        "engineer",
                        "__run__",
                    )
                    .expect("claim lookup")
                    .expect("singleton remains occupied")
            };
            assert_eq!(
                claim().agent_id,
                "planning-engineer",
                "Preparing intent never transfers ownership"
            );
            assert!(
                service
                    .prepare_singleton_publication(&record, 208, Some(&source))
                    .is_err(),
                "intent replay denied"
            );
            assert!(
                service
                    .store
                    .publish_review_replacement_prepared(&prepared, &mut registry)
                    .is_err(),
                "an unactivated replacement cannot publish or transfer"
            );
            assert_eq!(claim().agent_id, "planning-engineer");
            if finish_activation {
                // Typed activation fixtures exercise only the cross-resource
                // seam; the broker integration separately proves real peers.
                replacement.claim(207, [10; 32]).expect("fixture claim");
                replacement
                    .activate(208, [10; 32])
                    .expect("fixture activation");
                std::fs::write(
                    fixture.join(format!(
                        "runs/v645/dispatch/pending-{}.json",
                        hex(&replacement.launch_id_hash)
                    )),
                    serde_json::to_vec(&replacement).unwrap(),
                )
                .expect("active pending fixture");
                std::fs::write(
                    fixture.join("runs/v645/dispatch/replacement-engineer.json"),
                    &prepared.record_json,
                )
                .expect("active record fixture");
                let reconciled = service
                    .store
                    .reconcile_singletons(&mut registry, 209)
                    .expect("finish interrupted activation publication");
                assert_eq!(reconciled.published, 1);
                assert_eq!(claim().agent_id, "replacement-engineer");
                assert_eq!(
                    claim().publication_nonce.as_deref(),
                    Some(prepared.nonce.as_str())
                );
                assert!(
                    service
                        .store
                        .publish_review_replacement_prepared(&prepared, &mut registry)
                        .is_err(),
                    "transfer replay denied"
                );
                assert_eq!(
                    service
                        .store
                        .reconcile_singletons(&mut registry, 210)
                        .expect("idempotent reconcile")
                        .unchanged,
                    2
                );
            } else {
                let reconciled = service
                    .store
                    .reconcile_singletons(&mut registry, 209)
                    .expect("quarantine unlaunched intent");
                assert_eq!(reconciled.quarantined, 1);
                assert_eq!(
                    claim().agent_id,
                    "planning-engineer",
                    "failed intent does not release old singleton"
                );
            }
            assert_eq!(
                registry
                    .load_dispatch_publication(&old.nonce)
                    .unwrap()
                    .unwrap(),
                old,
                "old terminal history remains exact"
            );
            assert_eq!(
                service
                    .store
                    .load_for_run(&subject.run, &subject.expected_attachment.agent_id)
                    .unwrap()
                    .state,
                DispatchState::Malignant
            );
            // Close the store's handles first. Windows refuses to remove a
            // directory whose files are still open, and the bounded retry in
            // remove_fixture exists to surface exactly this leak rather than
            // hide it. Unix unlinks open files, so only Windows failed.
            drop(service);
            drop(registry);
            remove_fixture(&fixture);
        }
    }

    #[test]
    fn planning_review_terminal_recovery_rejects_forged_or_noncanonical_deltas() {
        for change in [
            "metadata",
            "pending-scope",
            "active-custody",
            "noncanonical-bytes",
        ] {
            let (fixture, service, subject, old) = published_planning_review_fixture();
            let mut interrupted = service.clone();
            interrupted.registry_path = None;
            for round in 1..=4 {
                interrupted
                    .review_ruling(planning_review_ruling(round), 200 + i64::from(round))
                    .expect("Native ruling");
            }
            let snapshot = service
                .store
                .read_review_terminal_snapshot(&subject.run, &subject.expected_attachment.agent_id)
                .unwrap();
            let record_path = fixture.join("runs/v645/dispatch/planning-engineer.json");
            match change {
                "metadata" => {
                    let mut record = snapshot.subject;
                    record.model = Some("forged-model".into());
                    std::fs::write(
                        &record_path,
                        format!("{}\n", serde_json::to_string(&record).unwrap()),
                    )
                    .unwrap();
                }
                "pending-scope" => {
                    let mut pending = snapshot.pending;
                    pending
                        .write_scope
                        .push(PathAuthority::exact(".shepherd/runs/v645/plan.md").unwrap());
                    std::fs::write(
                        fixture.join(format!(
                            "runs/v645/dispatch/pending-{}.json",
                            hex(&pending.launch_id_hash)
                        )),
                        serde_json::to_vec(&pending).unwrap(),
                    )
                    .unwrap();
                }
                "active-custody" => {
                    let mut custody = ReviewCustody::begin(
                        &snapshot.custody.rulings[0],
                        snapshot.custody.root_session_id.clone(),
                        snapshot.custody.subject_session_id.clone(),
                        snapshot.custody.subject_role,
                        None,
                        subject.launch_id_hash,
                    )
                    .unwrap();
                    for ruling in &snapshot.custody.rulings[..3] {
                        custody.apply(ruling).unwrap();
                    }
                    std::fs::write(
                        fixture.join("runs/v645/dispatch/.review-custody.planning-engineer.json"),
                        serde_json::to_vec(&custody).unwrap(),
                    )
                    .unwrap();
                }
                "noncanonical-bytes" => {
                    let mut bytes = std::fs::read(&record_path).unwrap();
                    bytes.push(b' ');
                    std::fs::write(&record_path, bytes).unwrap();
                }
                _ => unreachable!(),
            }
            let mut registry =
                Registry::open_migrated(service.registry_path.as_ref().unwrap()).unwrap();
            let report = service
                .store
                .reconcile_singletons(&mut registry, 205)
                .expect("reject forged delta");
            assert_eq!(report.refreshed, 0, "must not bless {change}");
            assert_eq!(
                report.quarantined, 1,
                "reject {change} instead of accepting generic changed bytes"
            );
            let retained = registry
                .load_dispatch_publication(&old.nonce)
                .unwrap()
                .unwrap();
            assert_eq!(
                retained.record_json, old.record_json,
                "never replace receipt with forged evidence"
            );
            // Close the store's handles first. Windows refuses to remove a
            // directory whose files are still open, and the bounded retry in
            // remove_fixture exists to surface exactly this leak rather than
            // hide it. Unix unlinks open files, so only Windows failed.
            drop(service);
            drop(registry);
            remove_fixture(&fixture);
        }
    }

    #[test]
    fn planning_review_rejects_wrong_mode_phase_lane_role_and_critic_ancestry() {
        for (mode, status) in [
            ("execution", "executing"),
            ("execution", "planted"),
            ("planning", "planned"),
        ] {
            let (fixture, service, _, _) = planning_review_fixture(mode, status);
            assert!(
                service
                    .review_ruling(planning_review_ruling(1), 201)
                    .is_err(),
                "run-scoped review requires planning/planted, not {mode}/{status}"
            );
            // Close the store's handles first. Windows refuses to remove a
            // directory whose files are still open, and the bounded retry in
            // remove_fixture exists to surface exactly this leak rather than
            // hide it. Unix unlinks open files, so only Windows failed.
            drop(service);
            remove_fixture(&fixture);
        }
        let (fixture, service, subject, reviewer) = planning_review_fixture("planning", "planted");
        let subject_record = active_review_record(&subject);
        let reviewer_record = active_review_record(&reviewer);
        let mut fabricated_lane = planning_review_ruling(1);
        fabricated_lane.review.lane = Some(LaneId::new("planning").expect("sentinel lane"));
        assert!(
            service.review_ruling(fabricated_lane, 201).is_err(),
            "no invented planning lane"
        );
        for (label, record, changed) in [
            ("unrelated-critic", &reviewer_record, {
                let mut r = reviewer_record.clone();
                r.parent_agent_id = None;
                r
            }),
            ("critic-lane", &reviewer_record, {
                let mut r = reviewer_record.clone();
                r.lane = Some(LaneId::new("lane-a").unwrap());
                r
            }),
            ("subject-role", &subject_record, {
                let mut r = subject_record.clone();
                r.role = Role::Coder;
                r
            }),
            ("subject-lane", &subject_record, {
                let mut r = subject_record.clone();
                r.lane = Some(LaneId::new("lane-a").unwrap());
                r
            }),
        ] {
            let path = fixture.join(format!("runs/v645/dispatch/{}.json", record.agent_id));
            std::fs::write(&path, serde_json::to_vec(&changed).expect("negative JSON"))
                .expect("negative fixture");
            assert!(
                service
                    .review_ruling(planning_review_ruling(1), 201)
                    .is_err(),
                "deny {label}"
            );
            std::fs::write(path, serde_json::to_vec(record).expect("original JSON"))
                .expect("restore fixture");
        }
        assert!(
            service
                .store
                .load_review_custody(&subject.run, &subject_record.agent_id)
                .is_err(),
            "invalid reviews must not create custody"
        );
        service
            .review_ruling(planning_review_ruling(1), 201)
            .expect("exact live Engineer and its descendant Critic are accepted");
        // Close the store's handles first. Windows refuses to remove a
        // directory whose files are still open, and the bounded retry in
        // remove_fixture exists to surface exactly this leak rather than
        // hide it. Unix unlinks open files, so only Windows failed.
        drop(service);
        remove_fixture(&fixture);
    }

    #[test]
    fn planning_review_fourth_rejection_preserves_identity_and_root_replacement_contract() {
        let (fixture, service, subject, _) = planning_review_fixture("planning", "planted");
        let run = &subject.run;
        let agent = &subject.expected_attachment.agent_id;
        let original = service.store.load_for_run(run, agent).expect("Engineer");
        let replacement_request = |agent: &str| ReviewReplacementRequest {
            schema: REVIEW_REPLACEMENT_REQUEST_SCHEMA.into(),
            run: "v645".into(),
            harness: Harness::Codex,
            root_session_id: "review-root".into(),
            subject_agent_id: "planning-engineer".into(),
            replacement_agent_id: agent.into(),
        };
        for round in 1..=3 {
            let custody = service
                .review_ruling(planning_review_ruling(round), 200 + i64::from(round))
                .expect("same Engineer receives bounded Critic RED");
            assert_eq!(custody.state, ReviewCustodyState::Active);
            assert_eq!(custody.rejected_revisions, round);
            assert_eq!(custody.rulings.len(), usize::from(round));
            assert_eq!(custody.task_generation, 1);
            assert!(custody.lane.is_none());
            assert!(
                !custody.claim_revoked && !custody.write_revoked && !custody.session_quarantined
            );
            assert_eq!(
                service
                    .store
                    .load_for_run(run, agent)
                    .expect("same Engineer"),
                original
            );
            assert_eq!(
                service
                    .store
                    .load_pending_for_agent(run, agent)
                    .expect("same pending"),
                subject
            );
        }
        let mut changed_generation = planning_review_ruling(4);
        changed_generation.task_generation = 2;
        assert!(
            service.review_ruling(changed_generation, 204).is_err(),
            "cannot reset review budget by changing generation"
        );
        assert!(
            service
                .review_replace(replacement_request("unprepared-replacement"), 204)
                .is_err(),
            "root cannot replace a still-active Engineer before rejection four"
        );
        let terminal = service
            .review_ruling(planning_review_ruling(4), 204)
            .expect("fourth rejection");
        assert_eq!(terminal.state, ReviewCustodyState::Malignant);
        assert_eq!(terminal.rulings.len(), 4);
        assert!(terminal.claim_revoked && terminal.write_revoked && terminal.session_quarantined);
        assert_eq!(
            service
                .store
                .load_for_run(run, agent)
                .expect("terminal Engineer")
                .state,
            DispatchState::Malignant
        );
        assert_eq!(
            service
                .store
                .load_pending_for_agent(run, agent)
                .expect("quarantined pending")
                .launch_state,
            PendingLaunchState::Quarantined
        );
        assert!(
            service
                .review_ruling(planning_review_ruling(5), 205)
                .is_err(),
            "terminal review cannot replay"
        );
        service
            .review_verify_terminal(
                ReviewTerminalVerificationRequest {
                    schema: "shepherd.review-terminal-verification-request/1".into(),
                    run: "v645".into(),
                    harness: Harness::Codex,
                    root_session_id: "review-root".into(),
                    subject_agent_id: agent.to_string(),
                    subject_session_id: subject.expected_child_session_id.to_string(),
                    task_generation: 1,
                    task_sha256: hex(&subject.task_sha256),
                    pending_launch_id_hash: hex(&subject.launch_id_hash),
                },
                205,
            )
            .expect("malignant planning Engineer has native prelaunch predicate denials");

        let mut drifted =
            planning_review_pending("drifted-engineer", 33, false, Some("planning-engineer"));
        drifted
            .write_scope
            .push(PathAuthority::exact(".shepherd/runs/v645/plan.md").expect("drifted scope"));
        service
            .store
            .publish_pending(&drifted)
            .expect("negative replacement fixture");
        assert!(
            service
                .review_replace(replacement_request("drifted-engineer"), 206)
                .is_err(),
            "planning root replacement cannot enlarge scope"
        );
        let exact =
            planning_review_pending("replacement-engineer", 34, false, Some("planning-engineer"));
        service
            .store
            .publish_pending(&exact)
            .expect("exact replacement fixture");
        let mut state = service.store.load_run(run).expect("run");
        state.status = "planned".into();
        state
            .store(&fixture.join("runs/v645/run.json"))
            .expect("closed planning fixture");
        assert!(
            service
                .review_replace(replacement_request("replacement-engineer"), 207)
                .is_err(),
            "planning replacement is not permitted after the planning phase"
        );
        state.status = "planted".into();
        state
            .store(&fixture.join("runs/v645/run.json"))
            .expect("restore planted fixture");
        let replaced = service.review_replace(replacement_request("replacement-engineer"), 208)
            .expect("exact planning root replaces its quarantined Engineer under unchanged task and scope");
        assert_eq!(replaced.state, ReviewCustodyState::Replaced);
        assert_eq!(replaced.rulings, terminal.rulings);
        assert_eq!(replaced.task_generation, 1);
        assert_eq!(
            replaced.replacement_agent_id.as_ref().map(AgentId::as_str),
            Some("replacement-engineer")
        );
        assert!(replaced.claim_revoked && replaced.write_revoked && replaced.session_quarantined);
        assert_eq!(
            service
                .store
                .load_for_run(run, agent)
                .expect("old Engineer stays terminal")
                .state,
            DispatchState::Malignant
        );
        assert!(
            service
                .review_ruling(planning_review_ruling(6), 209)
                .is_err(),
            "replacement never revives old lineage"
        );
        // Close the store's handles first. Windows refuses to remove a
        // directory whose files are still open, and the bounded retry in
        // remove_fixture exists to surface exactly this leak rather than
        // hide it. Unix unlinks open files, so only Windows failed.
        drop(service);
        remove_fixture(&fixture);
    }

    #[test]
    fn review_service_quarantines_rejection_four_and_allows_only_exact_root_lineage() {
        let fixture = std::env::temp_dir().join(format!(
            "shepherd-review-custody-{}-{}",
            std::process::id(),
            crate::dispatch_broker::now_millis()
        ));
        std::fs::create_dir_all(&fixture).expect("fixture");
        let fixture = std::fs::canonicalize(fixture).expect("canonical fixture");
        let runs = fixture.join("runs");
        let run = RunId::new("v645").expect("run");
        let state: shepherd::RunState = serde_json::from_value(serde_json::json!({
            "run": "v645",
            "status": "executing",
        }))
        .expect("run state");
        state
            .store(&runs.join("v645/run.json"))
            .expect("persist run");
        let project_id = ProjectId::new("018f47ce-72d7-7f64-9eb1-2f651d521c2a").expect("project");
        let store = DispatchStore::new(&runs);
        let binding = RootSessionBinding {
            schema: ROOT_SESSION_SCHEMA.into(),
            project_id: project_id.clone(),
            run: run.clone(),
            harness: Harness::Codex,
            session_id: SessionId::new("review-root").expect("root"),
            role: Role::Shepherd,
            mode: "execution".into(),
            project_filesystem_id: None,
            bound_at: 1,
            expires_at: 10_000,
        };
        store
            .publish_root_binding_for_run(&binding)
            .expect("root binding");
        store
            .activate_current_root_binding(&binding)
            .expect("current root");

        let mut subject = review_pending(
            "subject-agent",
            "subject-session",
            1,
            None,
            ".shepherd/runs/v645/lanes/lane-a/handoff.md",
        );
        subject.claim(100, [10; 32]).expect("claim subject");
        subject.activate(101, [10; 32]).expect("activate subject");
        let subject_record = active_review_record(&subject);
        let mut completed_control = subject_record.clone();
        completed_control
            .stop(StopRequest {
                agent_id: subject_record.agent_id.clone(),
                expected_revision: subject_record.revision,
                stopped_at: 150,
                result_artifact: subject_record.result_artifact.clone(),
            })
            .expect("ordinary completion control");
        assert_eq!(
            completed_control
                .resume(
                    review_resume_input(&completed_control, 160)
                        .expect("valid unpublished resume arguments")
                )
                .expect("nonmalignant resume is possible")
                .state,
            DispatchState::Active
        );
        let mut pending_control = review_pending(
            "claim-control",
            "claim-session",
            14,
            None,
            ".shepherd/runs/v645/lanes/lane-a/handoff.md",
        );
        pending_control
            .claim(100, [10; 32])
            .expect("unconsumed claim is possible");
        store.publish_pending(&subject).expect("subject pending");
        std::fs::write(
            runs.join("v645/dispatch/subject-agent.json"),
            serde_json::to_vec(&subject_record).expect("subject json"),
        )
        .expect("subject record");
        let reviewer = auditor_record();
        std::fs::write(
            runs.join("v645/dispatch/auditor-agent.json"),
            serde_json::to_vec(&reviewer).expect("reviewer json"),
        )
        .expect("reviewer record");
        let service = DispatchService::new(store, project_id);

        let ruling = |marker: u8| ReviewRulingRequest {
            schema: REVIEW_RULING_REQUEST_SCHEMA.into(),
            run: "v645".into(),
            harness: Harness::Codex,
            root_session_id: "review-root".into(),
            subject_agent_id: "subject-agent".into(),
            reviewer_dispatch_id: "auditor-agent".into(),
            reviewer_session_id: "auditor-session".into(),
            task_generation: 1,
            review: redo_review(marker),
        };
        let terminal_request = serde_json::json!({
            "schema": "shepherd.review-terminal-verification-request/1",
            "run": "v645", "harness": "codex", "root_session_id": "review-root",
            "subject_agent_id": "subject-agent", "subject_session_id": "subject-session",
            "task_generation": 1, "task_sha256": hex(&subject.task_sha256),
            "pending_launch_id_hash": hex(&subject.launch_id_hash),
        });
        for marker in 1..=3 {
            let custody = service
                .review_ruling(ruling(marker), 200 + i64::from(marker))
                .expect("bounded rejection");
            assert_eq!(custody.state, ReviewCustodyState::Active);
            assert_eq!(custody.rejected_revisions, marker);
            validate_review_claim_custody(&custody)
                .expect("active custody permits the native claim predicate");
            assert_eq!(
                service
                    .store
                    .load_for_run(&run, &subject_record.agent_id)
                    .expect("active subject")
                    .state,
                DispatchState::Active
            );
            assert_eq!(
                service
                    .store
                    .load_pending(&run, subject.launch_id_hash)
                    .expect("active pending")
                    .launch_state,
                PendingLaunchState::Active
            );
        }
        assert!(
            service
                .review_verify_terminal(
                    serde_json::from_value(terminal_request.clone()).expect("terminal request"),
                    204,
                )
                .is_err(),
            "three redos are not terminal proof"
        );
        let custody = service
            .review_ruling(ruling(4), 204)
            .expect("fourth rejection terminalizes atomically");
        assert_eq!(custody.state, ReviewCustodyState::Malignant);
        assert!(custody.claim_revoked && custody.write_revoked && custody.session_quarantined);
        assert_eq!(
            service
                .store
                .load_for_run(&run, &subject_record.agent_id)
                .expect("malignant subject")
                .state,
            DispatchState::Malignant
        );
        assert_eq!(
            service
                .store
                .load_pending(&run, subject.launch_id_hash)
                .expect("quarantined pending")
                .launch_state,
            PendingLaunchState::Quarantined
        );
        assert!(
            service.review_ruling(ruling(5), 205).is_err(),
            "review replay cannot revive or re-rerule the destroyed identity"
        );
        let read_dispatch_bytes = || {
            std::fs::read_dir(runs.join("v645/dispatch"))
                .expect("dispatch entries")
                .map(|entry| {
                    let path = entry.expect("entry").path();
                    let bytes = std::fs::read(&path).expect("dispatch bytes");
                    (path, bytes)
                })
                .collect::<std::collections::BTreeMap<_, _>>()
        };
        let before_verification = read_dispatch_bytes();
        let proof = service
            .review_verify_terminal(
                serde_json::from_value(terminal_request.clone()).expect("terminal request"),
                205,
            )
            .expect("native read-only terminal predicates");
        assert_eq!(proof["schema"], "shepherd.review-terminal-verification/1");
        assert_eq!(proof["proof_kind"], "native-prelaunch-predicate-denials");
        assert_eq!(proof["broker_peer_attempted"], false);
        assert_eq!(proof["provider_launch_attempted"], false);
        assert_eq!(proof["subject_mutated"], false);
        assert_eq!(
            proof["denials"]["resume"],
            DispatchError::ReviewCustodyTerminal.to_string()
        );
        assert_eq!(
            proof["denials"]["reclaim"],
            DispatchError::ReviewCustodyTerminal.to_string()
        );
        assert_eq!(
            proof["pending_claim_denial"],
            DispatchError::PendingLaunchConsumed.to_string()
        );
        assert_eq!(
            proof["denials"]["replay"],
            DispatchError::ReviewCustodyTerminal.to_string()
        );
        assert_eq!(
            read_dispatch_bytes(),
            before_verification,
            "verification must not repair or rewrite custody"
        );
        for (field, value) in [
            ("root_session_id", serde_json::json!("other-root")),
            ("subject_session_id", serde_json::json!("other-session")),
            ("harness", serde_json::json!("claude")),
            ("task_generation", serde_json::json!(2)),
            ("task_sha256", serde_json::json!("a".repeat(64))),
            ("pending_launch_id_hash", serde_json::json!("a".repeat(64))),
        ] {
            let mut forged = terminal_request.clone();
            forged[field] = value;
            assert!(
                service
                    .review_verify_terminal(
                        serde_json::from_value(forged).expect("forged request"),
                        205,
                    )
                    .is_err(),
                "mismatched {field} cannot produce proof"
            );
        }
        assert!(
            service
                .review_verify_terminal(
                    serde_json::from_value(terminal_request.clone()).expect("terminal request"),
                    10_001,
                )
                .is_err(),
            "an expired lease must not masquerade as terminal denial"
        );
        assert_eq!(read_dispatch_bytes(), before_verification);
        for (path, inconsistent) in [
            (
                runs.join("v645/dispatch/subject-agent.json"),
                serde_json::to_vec(&subject_record).expect("active subject"),
            ),
            (
                runs.join(format!(
                    "v645/dispatch/pending-{}.json",
                    hex(&subject.launch_id_hash)
                )),
                serde_json::to_vec(&subject).expect("active pending"),
            ),
        ] {
            std::fs::write(&path, inconsistent).expect("partial quarantine fixture");
            let before_denial = read_dispatch_bytes();
            assert!(
                service
                    .review_verify_terminal(
                        serde_json::from_value(terminal_request.clone()).expect("terminal request"),
                        205,
                    )
                    .is_err(),
                "partial/forged persisted quarantine must not produce proof"
            );
            assert_eq!(
                read_dispatch_bytes(),
                before_denial,
                "read-only verifier must not run normal-loader recovery"
            );
            std::fs::write(
                &path,
                before_verification
                    .get(&path)
                    .expect("original terminal bytes"),
            )
            .expect("restore unit fixture");
        }

        let mut non_root = review_pending(
            "non-root-replacement",
            "non-root-session",
            11,
            Some("subject-agent"),
            ".shepherd/runs/v645/lanes/lane-a/handoff.md",
        );
        non_root.caller_role = Role::Conductor;
        non_root.parent_dispatch_id = Some(DispatchId::new("conductor-parent").expect("parent"));
        non_root.role = Role::Worker;
        non_root.work_kind = WorkKind::Artifact;
        non_root.expected_attachment.role = Role::Worker;
        non_root.expected_attachment.startup_skill = "working".into();
        let non_root_path = runs.join(format!(
            "v645/dispatch/pending-{}.json",
            hex(&non_root.launch_id_hash)
        ));
        std::fs::write(
            &non_root_path,
            serde_json::to_vec(&non_root).expect("non-root json"),
        )
        .expect("non-root pending fixture");
        assert!(
            service
                .review_replace(
                    ReviewReplacementRequest {
                        schema: REVIEW_REPLACEMENT_REQUEST_SCHEMA.into(),
                        run: "v645".into(),
                        harness: Harness::Codex,
                        root_session_id: "review-root".into(),
                        subject_agent_id: "subject-agent".into(),
                        replacement_agent_id: "non-root-replacement".into(),
                    },
                    206,
                )
                .is_err(),
            "a non-root replacement claim fails closed at the service boundary"
        );
        std::fs::remove_file(non_root_path).expect("remove invalid negative fixture");

        let drifted = review_pending(
            "drifted-replacement",
            "drifted-session",
            12,
            Some("subject-agent"),
            ".shepherd/runs/v645/lanes/lane-a/reports/drifted.json",
        );
        service
            .store
            .publish_pending(&drifted)
            .expect("drifted pending");
        assert!(
            service
                .review_replace(
                    ReviewReplacementRequest {
                        schema: REVIEW_REPLACEMENT_REQUEST_SCHEMA.into(),
                        run: "v645".into(),
                        harness: Harness::Codex,
                        root_session_id: "review-root".into(),
                        subject_agent_id: "subject-agent".into(),
                        replacement_agent_id: "drifted-replacement".into(),
                    },
                    207,
                )
                .is_err(),
            "root cannot drift the destroyed task or scope"
        );

        let exact = review_pending(
            "exact-replacement",
            "exact-session",
            13,
            Some("subject-agent"),
            ".shepherd/runs/v645/lanes/lane-a/handoff.md",
        );
        service
            .store
            .publish_pending(&exact)
            .expect("exact replacement pending");
        let replaced = service
            .review_replace(
                ReviewReplacementRequest {
                    schema: REVIEW_REPLACEMENT_REQUEST_SCHEMA.into(),
                    run: "v645".into(),
                    harness: Harness::Codex,
                    root_session_id: "review-root".into(),
                    subject_agent_id: "subject-agent".into(),
                    replacement_agent_id: "exact-replacement".into(),
                },
                208,
            )
            .expect("exact root-owned lineage replacement");
        assert_eq!(replaced.state, ReviewCustodyState::Replaced);
        assert_eq!(
            replaced.replacement_agent_id.as_ref().map(AgentId::as_str),
            Some("exact-replacement")
        );
        // Close the store's handles first. Windows refuses to remove a
        // directory whose files are still open, and the bounded retry in
        // remove_fixture exists to surface exactly this leak rather than
        // hide it. Unix unlinks open files, so only Windows failed.
        drop(service);
        remove_fixture(&fixture);
    }

    #[test]
    fn planning_child_resolution_uses_its_exact_run_and_ignores_stale_siblings() {
        let fixture = std::env::temp_dir().join(format!(
            "shepherd-planning-child-resolution-{}",
            std::process::id(),
        ));
        std::fs::create_dir_all(&fixture).expect("fixture");
        let fixture = std::fs::canonicalize(fixture).expect("canonical fixture");
        let runs = fixture.join(".shepherd/runs");
        let mut state: shepherd::RunState = serde_json::from_value(serde_json::json!({
            "run": "v657", "status": "planted",
        }))
        .expect("run state");
        state
            .store(&runs.join("v657/run.json"))
            .expect("persist run state");
        let store = DispatchStore::new(&runs);
        let record = root_owned_role_record("v657", "planning-root", Role::Engineer);
        let root = RootSessionBinding {
            schema: ROOT_SESSION_SCHEMA.into(),
            project_id: record.project_id.clone(),
            run: record.run.clone(),
            harness: record.harness,
            session_id: record.root_session_id.clone(),
            role: Role::Shepherd,
            mode: "planning".into(),
            project_filesystem_id: None,
            bound_at: 10,
            expires_at: 10_000,
        };
        store
            .publish_root_binding_for_run(&root)
            .expect("planning root");
        store
            .activate_current_root_binding(&root)
            .expect("current root");
        std::fs::write(
            runs.join("v657/dispatch/engineer-v657.json"),
            serde_json::to_vec(&record).expect("record JSON"),
        )
        .expect("native child fixture");
        let service =
            DispatchService::with_project_root(store, record.project_id.clone(), &fixture);
        let request = ResolveDispatchRequest {
            schema: REQUEST_SCHEMA.into(),
            run: Some("v657".into()),
            harness: record.harness,
            agent_id: Some(record.agent_id.to_string()),
            agent_type: Some(record.agent_type.to_string()),
            role_carrier: Some(record.role.carrier()),
            lane: None,
            session_id: record.session_id.to_string(),
            tool_call_id: Some("planning-write".into()),
            tool_name: Some("Write".into()),
            tool_input: Some(serde_json::json!({"file_path": ".shepherd/runs/v657/plan.md"})),
        };
        for with_siblings in [false, true] {
            if with_siblings {
                std::fs::create_dir_all(runs.join("v512")).expect("historical run");
                std::fs::write(runs.join("v512/run.json"), b"{").expect("corrupt historical run");
                std::fs::create_dir_all(runs.join("v656")).expect("unrelated run");
                std::fs::write(
                    runs.join("v656/run.json"),
                    br#"{"run":"v656","status":"executing"}"#,
                )
                .expect("unrelated executing run");
            }
            let resolved = service
                .resolve(request.clone(), 101)
                .expect("planning child must use its exact run, not discover an executing sibling");
            assert_eq!(resolved.run, record.run);
            assert_eq!(resolved.role, Role::Engineer);
            assert_eq!(resolved.path_in_write_scope, Some(true));
        }
        let mut forged = request.clone();
        forged.session_id = "other-session".into();
        assert!(
            service.resolve(forged, 101).is_err(),
            "session mismatch must remain denied"
        );
        state.status = "closed".into();
        state
            .store(&runs.join("v657/run.json"))
            .expect("close selected run");
        assert!(
            service.resolve(request, 101).is_err(),
            "closed selected run must remain denied"
        );
        // Close the store's handles first. Windows refuses to remove a
        // directory whose files are still open, and the bounded retry in
        // remove_fixture exists to surface exactly this leak rather than
        // hide it. Unix unlinks open files, so only Windows failed.
        drop(service);
        remove_fixture(&fixture);
    }

    #[test]
    fn cross_run_root_continuation_revokes_prior_child_capabilities() {
        let fixture = std::env::temp_dir().join(format!(
            "shepherd-cross-run-child-revocation-{}",
            std::process::id()
        ));
        std::fs::create_dir_all(&fixture).expect("fixture");
        let fixture = std::fs::canonicalize(fixture).expect("canonical fixture");
        let runs = fixture.join("runs");
        for (run, status) in [("v645", "executing"), ("v657", "planted")] {
            let state: shepherd::RunState = serde_json::from_value(serde_json::json!({
                "run": run,
                "status": status,
            }))
            .expect("run state");
            state
                .store(&runs.join(run).join("run.json"))
                .expect("persist run state");
        }
        let store = DispatchStore::new(&runs);
        let project_id =
            ProjectId::new("018f47ce-72d7-7f64-9eb1-2f651d521c2a").expect("project id");
        let old = RootSessionBinding {
            schema: ROOT_SESSION_SCHEMA.into(),
            project_id: project_id.clone(),
            run: RunId::new("v645").expect("old run"),
            harness: Harness::Codex,
            session_id: SessionId::new("continuing-root").expect("root session"),
            role: Role::Shepherd,
            mode: "execution".into(),
            project_filesystem_id: None,
            bound_at: 10,
            expires_at: 10_000,
        };
        store
            .publish_root_binding_for_run(&old)
            .expect("old binding");
        store
            .activate_current_root_binding(&old)
            .expect("old current binding");
        let service = DispatchService::new(store, project_id);
        let old_record = root_owned_record("v645", "continuing-root");
        service
            .validate_dispatch_root_binding(&old_record, 101)
            .expect("old child initially belongs to current root");

        let mut next = old.clone();
        next.run = RunId::new("v657").expect("next run");
        next.mode = "planning".into();
        next.bound_at = 20;
        service
            .store
            .publish_root_binding_for_run(&next)
            .expect("next binding");
        service
            .store
            .activate_current_root_binding(&next)
            .expect("same root continues to v657");

        let error = service
            .validate_dispatch_root_binding(&old_record, 101)
            .expect_err("v645 child authority must end when the root continues to v657");
        assert!(error.to_string().contains("moved"), "{error}");
        service
            .validate_dispatch_root_binding(&root_owned_record("v657", "continuing-root"), 101)
            .expect("new-run child may bind to the continued root");
        // Close the store's handles first. Windows refuses to remove a
        // directory whose files are still open, and the bounded retry in
        // remove_fixture exists to surface exactly this leak rather than
        // hide it. Unix unlinks open files, so only Windows failed.
        drop(service);
        remove_fixture(&fixture);
    }

    #[test]
    fn unknown_markdown_is_not_an_implicit_non_code_allowlist() {
        let root = std::env::temp_dir().join(format!("shepherd-classify-{}", std::process::id()));
        std::fs::create_dir_all(root.join("docs")).expect("fixture");
        assert!(
            Command::new("git")
                .args(["init", "-q"])
                .current_dir(&root)
                .status()
                .expect("git init")
                .success()
        );
        let run = RunId::new("v645").expect("run");
        let scope = PathAuthority::new("random.md").expect("scope");
        assert_eq!(
            classify_scope(Some(&root), &scope, &run).expect("classification"),
            PathClass::Production
        );
        let production_doc = PathAuthority::new("crates/core/readme.md").expect("scope");
        assert_eq!(
            classify_scope(Some(&root), &production_doc, &run).expect("production classification"),
            PathClass::Production
        );
        let docs = PathAuthority::new("docs/**").expect("scope");
        assert_eq!(
            classify_scope(Some(&root), &docs, &run).expect("docs classification"),
            PathClass::NonCodeDeliverable
        );
        remove_fixture(&root);
    }

    #[test]
    fn role_scopes_bind_engineer_conductor_and_worker_to_exact_owned_artifacts() {
        let run = RunId::new("v657").expect("run");
        let lane = LaneId::new("lane-a").expect("lane");
        let result = PathAuthority::exact(".shepherd/runs/v657/lanes/lane-a/workers/worker.json")
            .expect("result");
        let review = PathAuthority::exact(".shepherd/runs/v657/lanes/lane-a/reviews/worker.json")
            .expect("review");
        let path = |value: &str| PathAuthority::new(value).expect("scope");

        for allowed in [
            ".shepherd/runs/v657/phase0.md",
            ".shepherd/runs/v657/plan.md",
            ".shepherd/runs/v657/graph/topology.json",
            ".shepherd/runs/v657/lanes/lane-b/plan.md",
            ".shepherd/runs/v657/reports/planning-correction.md",
        ] {
            assert!(
                role_scoped_path_allows(
                    Role::Engineer,
                    PathClass::RunArtifact,
                    &path(allowed),
                    &run,
                    None,
                    &result,
                ),
                "Engineer path {allowed}"
            );
        }
        for denied in [
            ".shepherd/runs/v657/seed.md",
            ".shepherd/runs/v657/graph/other.json",
            ".shepherd/runs/v657/lanes/lane-b/handoff.md",
            ".shepherd/runs/v657/lanes/lane-b/reviews/a.md",
            ".shepherd/runs/v657/lanes/lane-b/evidence/a.json",
            ".shepherd/runs/v657/lanes/*/plan.md",
        ] {
            assert!(
                !role_scoped_path_allows(
                    Role::Engineer,
                    PathClass::RunArtifact,
                    &path(denied),
                    &run,
                    None,
                    &result,
                ),
                "Engineer path {denied}"
            );
        }

        for allowed in [
            ".shepherd/runs/v657/lanes/lane-a/handoff.md",
            ".shepherd/runs/v657/lanes/lane-a/reports/gate.json",
            ".shepherd/runs/v657/lanes/lane-a/reviews/auditor.json",
            ".shepherd/runs/v657/lanes/lane-a/evidence/test.json",
        ] {
            assert!(
                role_scoped_path_allows(
                    Role::Conductor,
                    PathClass::RunArtifact,
                    &path(allowed),
                    &run,
                    Some(&lane),
                    &result,
                ),
                "Conductor path {allowed}"
            );
        }
        for denied in [
            ".shepherd/runs/v657/plan.md",
            ".shepherd/runs/v657/graph/topology.json",
            ".shepherd/runs/v657/lanes/lane-b/handoff.md",
            ".shepherd/runs/v657/lanes/lane-a2/handoff.md",
            ".shepherd/runs/v657/lanes/lane-a/not-reports/x.json",
            ".shepherd/runs/v657/lanes/lane-a/reports/nested/x.json",
            ".shepherd/runs/v657/lanes/lane-a/reports/*.json",
        ] {
            assert!(
                !role_scoped_path_allows(
                    Role::Conductor,
                    PathClass::RunArtifact,
                    &path(denied),
                    &run,
                    Some(&lane),
                    &result,
                ),
                "Conductor path {denied}"
            );
        }

        assert!(role_scoped_path_allows(
            Role::Worker,
            PathClass::NonCodeDeliverable,
            &path("docs/result.md"),
            &run,
            Some(&lane),
            &result,
        ));
        assert!(!role_scoped_path_allows(
            Role::Worker,
            PathClass::NonCodeDeliverable,
            &path("docs/**"),
            &run,
            Some(&lane),
            &result,
        ));
        assert!(role_scoped_path_allows(
            Role::Worker,
            PathClass::RunArtifact,
            &result,
            &run,
            Some(&lane),
            &result,
        ));
        assert!(!role_scoped_path_allows(
            Role::Worker,
            PathClass::RunArtifact,
            &path(".shepherd/runs/v657/plan.md"),
            &run,
            Some(&lane),
            &result,
        ));
        validate_role_artifact_namespaces(&run, Role::Worker, Some(&lane), &result, &review)
            .expect("canonical Worker artifact namespaces");
        let substituted_plan =
            PathAuthority::exact(".shepherd/runs/v657/plan.md").expect("substituted result");
        assert!(
            validate_role_artifact_namespaces(
                &run,
                Role::Worker,
                Some(&lane),
                &substituted_plan,
                &review,
            )
            .is_err(),
            "caller-selected result_artifact must not turn plan.md into Worker authority"
        );
        assert!(
            validate_role_artifact_namespaces(&run, Role::Worker, Some(&lane), &result, &result,)
                .is_err(),
            "Worker cannot substitute its result namespace for reviewer custody"
        );
    }

    #[test]
    fn repository_paths_preserve_exact_case_and_reject_unicode_and_alias_forms() {
        let root = std::env::temp_dir().join(format!("shepherd-path-{}", uuid::Uuid::now_v7()));
        std::fs::create_dir_all(root.join("Docs")).expect("fixture");
        for path in ["Cargo.toml", "Cargo.lock", "README.md", "Docs/task.md"] {
            std::fs::write(root.join(path), b"exact case\n").expect("case-preserving fixture");
        }
        let root = std::fs::canonicalize(root).expect("canonical fixture");
        for path in [
            "Cargo.toml",
            "Cargo.lock",
            "README.md",
            "Docs/task.md",
            "Docs/NewFile.md",
        ] {
            assert_eq!(
                repository_relative_path(Some(&root), path).expect("legitimate repository path"),
                path
            );
        }
        for path in [
            "cargo.toml",
            "docs/task.md",
            "Docs/Task.md",
            "docs/task.md ",
            "docs/task.md~",
            "docs/é.md",
            "C:/Docs/task.md",
            "C:\\Docs\\task.md",
            "Docs/task.md:stream",
            "Docs/../Cargo.toml",
        ] {
            assert!(
                repository_relative_path(Some(&root), path).is_err(),
                "unsafe repository path accepted: {path}"
            );
        }
        remove_fixture(&root);
    }

    #[test]
    fn trusted_generated_paths_accept_canonical_skill_names_but_reject_case_aliases() {
        let root =
            std::env::temp_dir().join(format!("shepherd-generated-path-{}", std::process::id()));
        std::fs::create_dir_all(root.join("skills/planning")).expect("fixture");
        std::fs::write(root.join("skills/planning/SKILL.md"), b"compiled skill\n")
            .expect("compiled skill");
        let root = std::fs::canonicalize(root).expect("canonical fixture");

        assert_eq!(
            trusted_manifest_relative_path(&root, "skills/planning/SKILL.md")
                .expect("compiler-owned exact spelling"),
            "skills/planning/SKILL.md"
        );
        assert!(
            trusted_manifest_relative_path(&root, "skills/planning/skill.md").is_err(),
            "case-folded alias must not inherit compiler authority"
        );
        assert_eq!(
            repository_relative_path(Some(&root), "skills/planning/SKILL.md")
                .expect("repository paths may use exact generated filenames"),
            "skills/planning/SKILL.md"
        );

        remove_fixture(&root);
    }

    #[test]
    fn native_scope_paths_preserve_case_without_reclassifying_native_state_aliases() {
        let root =
            std::env::temp_dir().join(format!("shepherd-scope-case-{}", uuid::Uuid::now_v7()));
        std::fs::create_dir_all(root.join(".shepherd")).expect("private Native directory");
        std::fs::write(root.join(".shepherd/project.json"), b"{}\n").expect("Native state");
        std::fs::write(root.join("Cargo.toml"), b"fixture\n").expect("production manifest");
        std::fs::write(root.join("README.md"), b"fixture\n").expect("non-code deliverable");
        let root = std::fs::canonicalize(root).expect("canonical fixture");
        assert!(
            Command::new("git")
                .args(["init", "-q"])
                .current_dir(&root)
                .status()
                .expect("fixture git")
                .success()
        );
        let run = RunId::new("v657").expect("run");
        let lane = LaneId::new("lane-a").expect("lane");
        let result = PathAuthority::exact(".shepherd/runs/v657/lanes/lane-a/reports/coder.md")
            .expect("result");
        let read = [PathAuthority::exact("README.md").expect("read scope")];
        let validate = |path: &str| {
            validate_native_scopes(
                Some(&root),
                NativeScopePolicy {
                    run: &run,
                    role: Role::Coder,
                    work_kind: WorkKind::ProductionCode,
                    lane: Some(&lane),
                    result_artifact: &result,
                },
                &read,
                &[PathAuthority::exact(path).expect("repository path syntax")],
            )
        };
        validate("Cargo.toml").expect("exact mixed-case production path");
        for path in [
            ".shepherd/project.json",
            ".SHEPHERD/project.json",
            ".shepherd/PROJECT.json",
            "cargo.toml",
        ] {
            assert!(
                validate(path).is_err(),
                "alias must not change Native class or path identity: {path}"
            );
        }
        validate_scope_nofollow(
            Some(&root),
            &PathAuthority::exact("NewDirectory/NewFile.rs").expect("new path"),
        )
        .expect("an absent unaliased path retains its normal missing-path classification");
        remove_fixture(&root);
    }

    #[test]
    fn installed_attachment_validation_accepts_each_canonical_package_layout() {
        for (target, harness, kind) in [
            ("claude", Harness::ClaudeCode, "claude-preload"),
            ("codex", Harness::Codex, "codex-custom-agent"),
            ("pi", Harness::Pi, "pi-skill-path"),
        ] {
            let root = std::env::temp_dir().join(format!(
                "shepherd-installed-layout-{}-{target}",
                std::process::id(),
            ));
            let compiler: crate::cmd::compile::CompileCmd =
                serde_json::from_value(serde_json::json!({
                    "target": target[..1].to_ascii_uppercase() + &target[1..],
                    "out": root,
                    "check": false,
                    "content_dir": null,
                }))
                .expect("canonical compile request");
            compiler.run().expect("materialize canonical package");
            let root = std::fs::canonicalize(root).expect("canonical package root");
            let binding =
                local_installed_package_binding(&root, &root.join(".shepherd-generated.json"))
                    .expect("retain installed package");
            let expected = trusted_attachment_expectation(
                &binding,
                CarrierAttachmentExpectationRequest {
                    target: harness,
                    role: "engineer".into(),
                    agent_id: "engineer-layout-test".into(),
                    attachment_kind: kind.into(),
                },
            )
            .expect("installed Engineer expectation");
            validate_attachment_files(&expected, &binding)
                .unwrap_or_else(|error| panic!("{target} package layout rejected: {error}"));
            let prefix = if target == "codex" {
                ".agents/skills"
            } else {
                "skills"
            };
            std::fs::write(root.join(prefix).join("planning/SKILL.md"), b"mutated\n")
                .expect("mutate installed startup skill");
            assert!(
                validate_attachment_files(&expected, &binding).is_err(),
                "{target} must reject changed installed skill bytes"
            );
            remove_fixture(&root);
        }
    }

    #[test]
    fn installed_startup_bundle_hashes_match_every_canonical_compiler_target() {
        use shepherd::compiler::{HarnessProfile, Target, compile};

        let input = crate::content_compiler::embedded_compile_input().expect("embedded content");
        for profile in HarnessProfile::canonical() {
            let tree = compile(&input, &profile).expect("compile canonical target");
            let root = std::env::temp_dir().join(format!(
                "shepherd-startup-bundle-{}-{}",
                std::process::id(),
                profile.target.as_str()
            ));
            std::fs::create_dir_all(&root).expect("fixture");
            let root = std::fs::canonicalize(root).expect("canonical fixture");
            for file in &tree.files {
                let path = root.join(&file.path);
                std::fs::create_dir_all(path.parent().expect("file parent"))
                    .expect("compiled directory");
                std::fs::write(&path, &file.content).expect("compiled file");
                #[cfg(unix)]
                {
                    use std::os::unix::fs::PermissionsExt;
                    std::fs::set_permissions(&path, std::fs::Permissions::from_mode(file.mode))
                        .expect("compiled file mode");
                }
            }
            let prefix = if profile.target == Target::Codex {
                ".agents/skills"
            } else {
                "skills"
            };
            for role in &tree.roles {
                let skill = role.startup_skill.as_deref().expect("startup skill");
                let expected = role
                    .startup_skill_sha256
                    .as_deref()
                    .expect("compiler startup digest");
                assert_eq!(
                    hex(&hash_skill_bundle(&root.join(prefix).join(skill), skill)
                        .expect("installed startup bundle")),
                    expected,
                    "{} {} startup bundle drifted from the compiler",
                    profile.target.as_str(),
                    role.role
                );
            }
            let planning = root.join(prefix).join("planning");
            let before = hash_skill_bundle(&planning, "planning").expect("original bundle");
            std::fs::write(planning.join("SKILL.md"), b"changed skill\n").expect("mutate bundle");
            assert_ne!(
                hash_skill_bundle(&planning, "planning").expect("changed bundle"),
                before,
                "changed installed bytes must invalidate the startup digest"
            );
            remove_fixture(&root);
        }
    }

    #[cfg(unix)]
    #[test]
    fn unix_metadata_normalization_accepts_signed_and_unsigned_values() {
        assert_eq!(metadata_u64(17_i64, "signed").expect("signed value"), 17);
        assert_eq!(
            metadata_u64(17_u64, "unsigned").expect("unsigned value"),
            17
        );
        assert!(
            metadata_u64(-1_i64, "negative").is_err(),
            "negative native metadata must not wrap"
        );
        assert!(
            metadata_u64(u128::from(u64::MAX) + 1, "wide").is_err(),
            "out-of-range native metadata must not truncate"
        );
    }

    #[cfg(unix)]
    #[test]
    #[allow(unsafe_code)]
    fn fifo_scope_is_rejected_without_opening_a_blocking_reader() {
        let root = std::env::temp_dir().join(format!("shepherd-fifo-scope-{}", std::process::id()));
        std::fs::create_dir_all(&root).expect("fixture");
        let root = std::fs::canonicalize(root).expect("canonical fixture");
        let fifo = root.join("fifo");
        let name = std::ffi::CString::new(fifo.as_os_str().as_bytes()).expect("fifo path");
        let result = unsafe { libc::mkfifo(name.as_ptr(), 0o600) };
        assert_eq!(result, 0, "mkfifo");
        let scope = PathAuthority::new("fifo").expect("scope");
        let error = validate_scope_nofollow(Some(&root), &scope)
            .expect_err("FIFO is not a safe regular path");
        assert!(
            error.to_string().contains("regular") || error.to_string().contains("special"),
            "{error}"
        );
        remove_fixture(&root);
    }
}