shepherd-core 6.6.0

//! Authenticated, single-use child-dispatch contracts.
//!
//! This module contains the host-neutral half of pending dispatch custody. It
//! validates the identity, edge, scope, task, lease, and carrier facts that
//! must agree before a native adapter may launch a child. Filesystem reads,
//! descriptor custody, and atomic persistence stay in the CLI crate.

use alloc::{collections::BTreeSet, format, string::String, vec::Vec};

use crate::Harness;
use crate::vocabulary::{RunStatus, Vocabulary};

use super::{
    AgentId, DispatchError, DispatchResult, LaneId, ProjectId, Role, RunId, SessionId,
    validate_write_scope_pattern,
};

pub const PENDING_DISPATCH_SCHEMA: &str = "shepherd.pending-dispatch/2";
pub const LOADED_CARRIER_SCHEMA: &str = "shepherd.loaded-carrier/1";

/// Compare fixed-size digest material without an early exit.
#[inline(never)]
pub fn constant_time_digest_eq(left: &[u8; 32], right: &[u8; 32]) -> bool {
    let mut difference = 0_u8;
    for index in 0..32 {
        difference |= left[index] ^ right[index];
    }
    difference == 0
}

pub(super) mod digest_serde {
    use alloc::{format, string::String};
    use serde::{Deserializer, Serialize, Serializer, de::Visitor};

    pub(crate) fn serialize<S>(value: &[u8; 32], serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        let mut text = String::with_capacity(64);
        for byte in value {
            text.push_str(&format!("{byte:02x}"));
        }
        text.serialize(serializer)
    }

    pub(crate) fn deserialize<'de, D>(deserializer: D) -> Result<[u8; 32], D::Error>
    where
        D: Deserializer<'de>,
    {
        struct DigestVisitor;

        impl<'de> Visitor<'de> for DigestVisitor {
            type Value = [u8; 32];

            fn expecting(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
                formatter.write_str("a 32-byte hexadecimal digest")
            }

            fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
            where
                E: serde::de::Error,
            {
                if value.len() != 64 {
                    return Err(E::custom("digest must contain 64 hexadecimal characters"));
                }
                let mut digest = [0; 32];
                let (pairs, remainder) = value.as_bytes().as_chunks::<2>();
                if !remainder.is_empty() {
                    return Err(E::custom("digest is not an even-length hexadecimal value"));
                }
                for (index, pair) in pairs.iter().enumerate() {
                    let high =
                        hex_digit(pair[0]).ok_or_else(|| E::custom("digest is not hexadecimal"))?;
                    let low =
                        hex_digit(pair[1]).ok_or_else(|| E::custom("digest is not hexadecimal"))?;
                    digest[index] = (high << 4) | low;
                }
                Ok(digest)
            }

            fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
            where
                A: serde::de::SeqAccess<'de>,
            {
                let mut digest = [0; 32];
                for (index, byte) in digest.iter_mut().enumerate() {
                    *byte = sequence.next_element()?.ok_or_else(|| {
                        serde::de::Error::custom(format!("digest ended at byte {index}"))
                    })?;
                }
                if sequence.next_element::<u8>()?.is_some() {
                    return Err(serde::de::Error::custom(
                        "digest contains more than 32 bytes",
                    ));
                }
                Ok(digest)
            }
        }

        deserializer.deserialize_any(DigestVisitor)
    }

    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,
        }
    }
}

pub(super) mod optional_digest_serde {
    use alloc::{format, string::String};
    use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error};

    pub(crate) fn serialize<S>(value: &Option<[u8; 32]>, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        value
            .map(|digest| {
                let mut text = String::with_capacity(64);
                for byte in digest {
                    text.push_str(&format!("{byte:02x}"));
                }
                text
            })
            .serialize(serializer)
    }

    pub(crate) fn deserialize<'de, D>(deserializer: D) -> Result<Option<[u8; 32]>, D::Error>
    where
        D: Deserializer<'de>,
    {
        let value = Option::<String>::deserialize(deserializer)?;
        value
            .map(|text| parse(&text).map_err(D::Error::custom))
            .transpose()
    }

    fn parse(value: &str) -> Result<[u8; 32], String> {
        if value.len() != 64 {
            return Err("optional digest must contain 64 hexadecimal characters".into());
        }
        let mut digest = [0_u8; 32];
        for (index, pair) in value.as_bytes().as_chunks::<2>().0.iter().enumerate() {
            let high =
                digit(pair[0]).ok_or_else(|| String::from("optional digest is not hexadecimal"))?;
            let low =
                digit(pair[1]).ok_or_else(|| String::from("optional digest is not hexadecimal"))?;
            digest[index] = high << 4 | low;
        }
        Ok(digest)
    }

    fn 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,
        }
    }
}

macro_rules! pending_id {
    ($name:ident, $kind:literal) => {
        #[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
        pub struct $name(String);

        impl $name {
            pub fn new(value: impl Into<String>) -> DispatchResult<Self> {
                let value = value.into();
                let bytes = value.as_bytes();
                if (1..=128).contains(&bytes.len())
                    && bytes[0].is_ascii_alphanumeric()
                    && bytes.iter().all(|byte| {
                        byte.is_ascii_alphanumeric() || matches!(*byte, b'.' | b'_' | b':' | b'-')
                    })
                    && value != "."
                    && value != ".."
                {
                    Ok(Self(value))
                } else {
                    Err(DispatchError::InvalidIdentifier { kind: $kind, value })
                }
            }

            #[must_use]
            pub fn as_str(&self) -> &str {
                &self.0
            }
        }

        impl core::fmt::Display for $name {
            fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
                formatter.write_str(&self.0)
            }
        }

        impl serde::Serialize for $name {
            fn serialize<S>(&self, serializer: S) -> core::result::Result<S::Ok, S::Error>
            where
                S: serde::Serializer,
            {
                serializer.serialize_str(&self.0)
            }
        }

        impl<'de> serde::Deserialize<'de> for $name {
            fn deserialize<D>(deserializer: D) -> core::result::Result<Self, D::Error>
            where
                D: serde::Deserializer<'de>,
            {
                let value = String::deserialize(deserializer)?;
                Self::new(value).map_err(serde::de::Error::custom)
            }
        }
    };
}

pending_id!(DispatchId, "dispatch id");

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct ProjectFilesystemId(String);

impl ProjectFilesystemId {
    pub fn new(value: impl Into<String>) -> DispatchResult<Self> {
        let value = value.into();
        if value.len() == 64 && value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
            Ok(Self(value.to_ascii_lowercase()))
        } else {
            Err(DispatchError::InvalidIdentifier {
                kind: "project filesystem id",
                value,
            })
        }
    }

    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl core::fmt::Display for ProjectFilesystemId {
    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        formatter.write_str(&self.0)
    }
}

impl serde::Serialize for ProjectFilesystemId {
    fn serialize<S>(&self, serializer: S) -> core::result::Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        serializer.serialize_str(&self.0)
    }
}

impl<'de> serde::Deserialize<'de> for ProjectFilesystemId {
    fn deserialize<D>(deserializer: D) -> core::result::Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        Self::new(String::deserialize(deserializer)?).map_err(serde::de::Error::custom)
    }
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct GitCommit(String);

impl GitCommit {
    pub fn new(value: impl Into<String>) -> DispatchResult<Self> {
        let value = value.into();
        if value.len() == 40 && value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
            Ok(Self(value.to_ascii_lowercase()))
        } else {
            Err(DispatchError::InvalidIdentifier {
                kind: "git commit",
                value,
            })
        }
    }

    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl core::fmt::Display for GitCommit {
    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        formatter.write_str(&self.0)
    }
}

impl serde::Serialize for GitCommit {
    fn serialize<S>(&self, serializer: S) -> core::result::Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        serializer.serialize_str(&self.0)
    }
}

impl<'de> serde::Deserialize<'de> for GitCommit {
    fn deserialize<D>(deserializer: D) -> core::result::Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        Self::new(String::deserialize(deserializer)?).map_err(serde::de::Error::custom)
    }
}

#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct PathAuthority(String);

impl PathAuthority {
    /// Construct a bounded repository-relative authority pattern.
    pub fn new(value: impl Into<String>) -> DispatchResult<Self> {
        let value = value.into();
        validate_write_scope_pattern(&value)?;
        if value == "*" || value == "**" {
            return Err(DispatchError::InvalidWriteScope(value));
        }
        Ok(Self(value))
    }

    /// Construct an authority for one exact repository-relative path.
    pub fn exact(value: impl Into<String>) -> DispatchResult<Self> {
        let value = value.into();
        if value.contains('*') {
            return Err(DispatchError::InvalidWriteScope(value));
        }
        Self::new(value)
    }

    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }

    #[must_use]
    pub fn is_exact(&self) -> bool {
        !self.0.contains('*')
    }

    pub fn contains(&self, path: &str) -> DispatchResult<bool> {
        super::path_in_write_scope(path, &alloc::vec![self.0.clone()])
    }
}

impl core::fmt::Display for PathAuthority {
    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        formatter.write_str(&self.0)
    }
}

impl serde::Serialize for PathAuthority {
    fn serialize<S>(&self, serializer: S) -> core::result::Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        serializer.serialize_str(&self.0)
    }
}

impl<'de> serde::Deserialize<'de> for PathAuthority {
    fn deserialize<D>(deserializer: D) -> core::result::Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        Self::new(String::deserialize(deserializer)?).map_err(serde::de::Error::custom)
    }
}

#[derive(
    Clone,
    Copy,
    Debug,
    Eq,
    Hash,
    PartialEq,
    serde::Deserialize,
    serde::Serialize,
    strum::AsRefStr,
    strum::Display,
    strum::EnumCount,
    strum::EnumIs,
    strum::EnumString,
    strum::IntoStaticStr,
    strum::VariantNames,
)]
#[serde(rename_all = "kebab-case")]
#[strum(ascii_case_insensitive, serialize_all = "snake_case")]
pub enum PendingLaunchState {
    Pending,
    ClaimedUnspawned,
    Active,
    Quarantined,
    LaunchFailed,
    Canceled,
    Expired,
}

impl PendingLaunchState {
    #[must_use]
    pub const fn can_claim(self) -> bool {
        matches!(self, Self::Pending)
    }

    #[must_use]
    pub const fn can_activate(self) -> bool {
        matches!(self, Self::ClaimedUnspawned)
    }

    #[must_use]
    pub const fn is_terminal(self) -> bool {
        matches!(
            self,
            Self::Quarantined | Self::LaunchFailed | Self::Canceled | Self::Expired
        )
    }
}

pub const LAUNCH_CLEANUP_SCHEMA: &str = "shepherd.launch-cleanup/1";

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct LaunchCleanupResponse {
    pub schema: String,
    #[serde(with = "digest_serde")]
    pub launch_id_hash: [u8; 32],
    pub state: PendingLaunchState,
}

impl LaunchCleanupResponse {
    pub fn validate(&self) -> DispatchResult<()> {
        if self.schema != LAUNCH_CLEANUP_SCHEMA {
            return Err(DispatchError::InvalidResponse(
                "unsupported launch cleanup schema".into(),
            ));
        }
        validate_digest(self.launch_id_hash, "launch identity")?;
        if !self.state.is_terminal() {
            return Err(DispatchError::InvalidResponse(
                "launch cleanup response is not terminal".into(),
            ));
        }
        Ok(())
    }
}

#[derive(
    Clone,
    Copy,
    Debug,
    Eq,
    Hash,
    PartialEq,
    serde::Deserialize,
    serde::Serialize,
    strum::AsRefStr,
    strum::Display,
    strum::EnumCount,
    strum::EnumIs,
    strum::EnumString,
    strum::IntoStaticStr,
    strum::VariantNames,
)]
#[serde(rename_all = "kebab-case")]
#[strum(ascii_case_insensitive, serialize_all = "snake_case")]
pub enum WorkKind {
    Planning,
    ProductionCode,
    Artifact,
    Review,
    Research,
    Coordination,
}

#[derive(
    Clone,
    Copy,
    Debug,
    Eq,
    Hash,
    PartialEq,
    serde::Deserialize,
    serde::Serialize,
    strum::AsRefStr,
    strum::Display,
    strum::EnumCount,
    strum::EnumIs,
    strum::EnumString,
    strum::IntoStaticStr,
    strum::VariantNames,
)]
#[serde(rename_all = "kebab-case")]
#[strum(ascii_case_insensitive, serialize_all = "snake_case")]
pub enum AttachmentKind {
    ClaudePreload,
    CodexCustomAgent,
    PiSkillPath,
}

impl AttachmentKind {
    pub fn validate_for(self, target: Harness) -> DispatchResult<()> {
        let valid = matches!(
            (target, self),
            (Harness::ClaudeCode, Self::ClaudePreload)
                | (Harness::Codex, Self::CodexCustomAgent)
                | (Harness::Pi, Self::PiSkillPath)
        );
        if valid {
            Ok(())
        } else {
            Err(DispatchError::AttachmentMismatch(format!(
                "attachment kind does not match target `{target}`"
            )))
        }
    }
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct CarrierAttachmentExpectation {
    pub target: Harness,
    pub role: Role,
    pub agent_id: AgentId,
    pub installed_carrier_path: String,
    #[serde(with = "digest_serde")]
    pub candidate_sha256: [u8; 32],
    #[serde(with = "digest_serde")]
    pub carrier_sha256: [u8; 32],
    #[serde(with = "digest_serde")]
    pub compiler_tree_sha256: [u8; 32],
    pub startup_skill: String,
    #[serde(with = "digest_serde")]
    pub skill_bundle_sha256: [u8; 32],
    pub attachment_kind: AttachmentKind,
}

impl CarrierAttachmentExpectation {
    pub fn validate(&self) -> DispatchResult<()> {
        validate_attachment_identity(
            self.target,
            self.role,
            &self.agent_id,
            &self.installed_carrier_path,
            &self.startup_skill,
            self.attachment_kind,
        )?;
        validate_digest(self.carrier_sha256, "carrier")?;
        validate_digest(self.candidate_sha256, "native candidate")?;
        validate_digest(self.compiler_tree_sha256, "compiler tree")?;
        validate_digest(self.skill_bundle_sha256, "skill bundle")
    }
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct LoadedCarrierAttestationV1 {
    pub schema: String,
    #[serde(with = "digest_serde")]
    pub nonce_sha256: [u8; 32],
    pub target: Harness,
    pub role: Role,
    pub agent_id: AgentId,
    pub installed_carrier_path: String,
    #[serde(with = "digest_serde")]
    pub candidate_sha256: [u8; 32],
    #[serde(with = "digest_serde")]
    pub carrier_sha256: [u8; 32],
    #[serde(with = "digest_serde")]
    pub compiler_tree_sha256: [u8; 32],
    pub startup_skill: String,
    #[serde(with = "digest_serde")]
    pub skill_bundle_sha256: [u8; 32],
    pub attachment_kind: AttachmentKind,
}

impl LoadedCarrierAttestationV1 {
    pub fn validate_against(
        &self,
        expected: &CarrierAttachmentExpectation,
        nonce_sha256: &[u8; 32],
    ) -> DispatchResult<()> {
        if self.schema != LOADED_CARRIER_SCHEMA {
            return Err(DispatchError::AttachmentMismatch(format!(
                "unsupported attestation schema `{}`",
                self.schema
            )));
        }
        expected.validate()?;
        validate_digest(self.nonce_sha256, "attachment nonce")?;
        if !constant_time_digest_eq(&self.nonce_sha256, nonce_sha256) {
            return Err(DispatchError::AttachmentMismatch(
                "attestation nonce does not match pending claim".into(),
            ));
        }
        if self.target != expected.target
            || self.role != expected.role
            || self.agent_id != expected.agent_id
            || self.installed_carrier_path != expected.installed_carrier_path
            || !constant_time_digest_eq(&self.candidate_sha256, &expected.candidate_sha256)
            || !constant_time_digest_eq(&self.carrier_sha256, &expected.carrier_sha256)
            || !constant_time_digest_eq(&self.compiler_tree_sha256, &expected.compiler_tree_sha256)
            || self.startup_skill != expected.startup_skill
            || !constant_time_digest_eq(&self.skill_bundle_sha256, &expected.skill_bundle_sha256)
            || self.attachment_kind != expected.attachment_kind
        {
            return Err(DispatchError::AttachmentMismatch(
                "attestation does not match the prepared carrier expectation".into(),
            ));
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct PendingDispatch {
    pub schema: String,
    #[serde(with = "digest_serde")]
    pub launch_id_hash: [u8; 32],
    #[serde(with = "digest_serde")]
    pub parent_process_hash: [u8; 32],
    pub project_id: ProjectId,
    pub project_filesystem_id: ProjectFilesystemId,
    pub run: RunId,
    pub run_status: Vocabulary<RunStatus>,
    pub root_session_id: SessionId,
    pub caller_role: Role,
    pub parent_dispatch_id: Option<DispatchId>,
    pub replaces_agent_id: Option<AgentId>,
    pub role: Role,
    pub work_kind: WorkKind,
    pub lane: Option<LaneId>,
    pub baseline_commit: GitCommit,
    pub read_scope: Vec<PathAuthority>,
    pub write_scope: Vec<PathAuthority>,
    pub result_artifact: PathAuthority,
    pub review_artifact: PathAuthority,
    pub task_path: PathAuthority,
    #[serde(with = "digest_serde")]
    pub task_sha256: [u8; 32],
    pub expected_child_session_id: SessionId,
    pub expected_attachment: CarrierAttachmentExpectation,
    pub expires_at: i64,
    pub launch_state: PendingLaunchState,
    pub claimed_at: Option<i64>,
    #[serde(with = "optional_digest_serde")]
    pub child_process_hash: Option<[u8; 32]>,
    pub activated_at: Option<i64>,
    #[serde(with = "digest_serde")]
    pub nonce_sha256: [u8; 32],
}

impl PendingDispatch {
    pub fn validate(&self) -> DispatchResult<()> {
        if self.schema != PENDING_DISPATCH_SCHEMA {
            return Err(DispatchError::InvalidPending(format!(
                "unsupported pending schema `{}`",
                self.schema
            )));
        }
        for (digest, label) in [
            (self.launch_id_hash, "launch identity"),
            (self.parent_process_hash, "parent process identity"),
            (self.task_sha256, "task"),
            (self.nonce_sha256, "pending nonce"),
        ] {
            validate_digest(digest, label)?;
        }
        if !self
            .run_status
            .known()
            .is_some_and(RunStatus::admits_dispatch)
        {
            return Err(DispatchError::InvalidPending(
                "pending dispatch run state is not dispatchable".into(),
            ));
        }
        if self.expires_at <= 0 {
            return Err(DispatchError::InvalidTime(
                "pending lease expiry must be positive".into(),
            ));
        }
        match self.launch_state {
            PendingLaunchState::Pending => {
                if self.claimed_at.is_some()
                    || self.child_process_hash.is_some()
                    || self.activated_at.is_some()
                {
                    return Err(DispatchError::InvalidPending(
                        "pending launch carries claimed state".into(),
                    ));
                }
            }
            PendingLaunchState::ClaimedUnspawned => {
                let Some(claimed_at) = self.claimed_at else {
                    return Err(DispatchError::InvalidPending(
                        "claimed launch has no claim time".into(),
                    ));
                };
                if claimed_at < 0 || claimed_at >= self.expires_at {
                    return Err(DispatchError::InvalidTime(
                        "pending claim time is outside the lease".into(),
                    ));
                }
                if self.child_process_hash.is_none() || self.activated_at.is_some() {
                    return Err(DispatchError::InvalidPending(
                        "claimed launch state is incomplete".into(),
                    ));
                }
            }
            PendingLaunchState::Active => {
                let (Some(claimed_at), Some(activated_at), Some(_child_process_hash)) =
                    (self.claimed_at, self.activated_at, self.child_process_hash)
                else {
                    return Err(DispatchError::InvalidPending(
                        "active launch state is incomplete".into(),
                    ));
                };
                if claimed_at < 0 || activated_at < claimed_at || activated_at >= self.expires_at {
                    return Err(DispatchError::InvalidTime(
                        "active launch timestamps are outside the lease".into(),
                    ));
                }
            }
            PendingLaunchState::Quarantined => {
                let (Some(claimed_at), Some(activated_at), Some(_child_process_hash)) =
                    (self.claimed_at, self.activated_at, self.child_process_hash)
                else {
                    return Err(DispatchError::InvalidPending(
                        "quarantined launch does not preserve its activated child identity".into(),
                    ));
                };
                if claimed_at < 0 || activated_at < claimed_at || activated_at >= self.expires_at {
                    return Err(DispatchError::InvalidTime(
                        "quarantined launch timestamps are outside the original lease".into(),
                    ));
                }
            }
            PendingLaunchState::LaunchFailed
            | PendingLaunchState::Canceled
            | PendingLaunchState::Expired => {
                if self.activated_at.is_some() {
                    return Err(DispatchError::InvalidPending(
                        "terminal launch state was already activated".into(),
                    ));
                }
            }
        }
        if self.read_scope.is_empty() {
            return Err(DispatchError::InvalidPending(
                "pending dispatch requires a bounded read scope".into(),
            ));
        }
        validate_unique_scopes(&self.read_scope, "read")?;
        validate_unique_scopes(&self.write_scope, "write")?;
        if !self.result_artifact.is_exact() || !self.review_artifact.is_exact() {
            return Err(DispatchError::InvalidArtifact(
                "result and review artifacts must be exact paths".into(),
            ));
        }
        if self.result_artifact == self.review_artifact {
            return Err(DispatchError::InvalidArtifact(
                "result and review artifacts must be distinct".into(),
            ));
        }
        if !self.task_path.is_exact() {
            return Err(DispatchError::InvalidPending(
                "task path must be exact".into(),
            ));
        }
        self.expected_attachment.validate()?;
        if self.expected_attachment.role != self.role
            || self.expected_attachment.agent_id.as_str().is_empty()
        {
            return Err(DispatchError::AttachmentMismatch(
                "expected attachment does not match the pending child".into(),
            ));
        }
        if self.parent_dispatch_id.is_none()
            && !matches!(self.caller_role, Role::Shepherd | Role::Planter)
        {
            return Err(DispatchError::PendingEdge(
                "non-root caller must carry parent dispatch ancestry".into(),
            ));
        }
        if self.parent_dispatch_id.is_some() && self.caller_role.is_root() {
            return Err(DispatchError::PendingEdge(
                "root caller cannot carry child dispatch ancestry".into(),
            ));
        }
        if self.replaces_agent_id.as_ref().is_some_and(|replaced| {
            self.caller_role != Role::Shepherd
                || self.parent_dispatch_id.is_some()
                || replaced == &self.expected_attachment.agent_id
        }) {
            return Err(DispatchError::PendingEdge(
                "replacement lineage requires a root-owned new child identity".into(),
            ));
        }
        validate_pending_edge(
            &self.run_status,
            self.caller_role,
            self.role,
            self.work_kind,
        )?;
        Ok(())
    }

    pub fn validate_edge(&self, run_status: &Vocabulary<RunStatus>) -> DispatchResult<()> {
        self.validate()?;
        validate_pending_edge(run_status, self.caller_role, self.role, self.work_kind)
    }

    pub fn check_lease(&self, now: i64) -> DispatchResult<()> {
        self.validate()?;
        if now < 0 || now >= self.expires_at {
            return Err(DispatchError::PendingExpired {
                expires_at: self.expires_at,
            });
        }
        if !self.launch_state.can_claim() {
            return Err(DispatchError::PendingLaunchConsumed);
        }
        Ok(())
    }

    pub fn claim(&mut self, now: i64, child_process_hash: [u8; 32]) -> DispatchResult<()> {
        self.check_lease(now)?;
        validate_digest(child_process_hash, "child process identity")?;
        self.launch_state = PendingLaunchState::ClaimedUnspawned;
        self.claimed_at = Some(now);
        self.child_process_hash = Some(child_process_hash);
        self.validate()
    }

    pub fn expire(&mut self) -> DispatchResult<()> {
        if self.launch_state == PendingLaunchState::Active {
            return Err(DispatchError::InvalidPending(
                "an active launch cannot be expired as unspawned".into(),
            ));
        }
        self.launch_state = PendingLaunchState::Expired;
        self.validate()
    }

    pub fn cancel(&mut self) -> DispatchResult<()> {
        if self.launch_state == PendingLaunchState::Active {
            return Err(DispatchError::InvalidPending(
                "an active launch cannot be canceled".into(),
            ));
        }
        self.launch_state = PendingLaunchState::Canceled;
        self.validate()
    }

    pub fn fail(&mut self) -> DispatchResult<()> {
        if self.launch_state == PendingLaunchState::Active {
            return Err(DispatchError::InvalidPending(
                "an active launch cannot be failed".into(),
            ));
        }
        self.launch_state = PendingLaunchState::LaunchFailed;
        self.validate()
    }

    /// Terminalize a launch after recovery found no durable active record.
    /// This is intentionally separate from ordinary cleanup: an active launch
    /// cannot be failed by a live caller, but a restart must not leave an
    /// active pending row orphaned when its record was never published.
    pub fn fail_after_recovery(&mut self) -> DispatchResult<()> {
        if !matches!(
            self.launch_state,
            PendingLaunchState::ClaimedUnspawned | PendingLaunchState::Active
        ) {
            return Err(DispatchError::InvalidPending(
                "recovery failure requires an in-flight launch".into(),
            ));
        }
        self.launch_state = PendingLaunchState::LaunchFailed;
        self.activated_at = None;
        self.validate()
    }

    /// Revoke a previously activated broker claim after Native review custody
    /// terminally quarantines its exact child identity.
    pub fn quarantine(&mut self) -> DispatchResult<()> {
        if self.launch_state != PendingLaunchState::Active {
            return Err(DispatchError::ReviewCustodyTerminal);
        }
        self.launch_state = PendingLaunchState::Quarantined;
        self.validate()
    }

    pub fn activate(&mut self, now: i64, child_process_hash: [u8; 32]) -> DispatchResult<()> {
        self.validate()?;
        if !self.launch_state.can_activate() {
            return Err(DispatchError::InvalidPending(
                "launch activation requires claimed_unspawned state".into(),
            ));
        }
        if self
            .child_process_hash
            .is_none_or(|claimed| !constant_time_digest_eq(&claimed, &child_process_hash))
        {
            return Err(DispatchError::InvalidPending(
                "activation process identity does not match the claimed child".into(),
            ));
        }
        if now < 0 || now >= self.expires_at {
            return Err(DispatchError::PendingExpired {
                expires_at: self.expires_at,
            });
        }
        self.launch_state = PendingLaunchState::Active;
        self.activated_at = Some(now);
        self.validate()
    }
}

pub fn validate_pending_edge(
    run_status: &Vocabulary<RunStatus>,
    caller: Role,
    target: Role,
    work_kind: WorkKind,
) -> DispatchResult<()> {
    if !target.allows_work_kind(work_kind) {
        return Err(DispatchError::PendingEdge(format!(
            "role `{target}` cannot receive work kind `{work_kind:?}`"
        )));
    }
    // Exhaustive over `RunStatus` on purpose: adding a status to the
    // vocabulary must not silently inherit "authorizes nothing" from a
    // catch-all. This match was keyed on `&str`, and had drifted to an arm for
    // `"integrating"` -- a status the run store rejects on both read and write,
    // so that arm had never once been taken.
    let allowed = match run_status.known() {
        Some(RunStatus::Planted) => matches!(
            (caller, target, work_kind),
            (Role::Shepherd, Role::Engineer, WorkKind::Planning)
                | (Role::Engineer, Role::Auditor, WorkKind::Review)
                | (Role::Engineer, Role::Discovery, WorkKind::Research)
                | (Role::Engineer, Role::Critic, WorkKind::Review)
        ),
        Some(RunStatus::Executing) => matches!(
            (caller, target, work_kind),
            (Role::Shepherd, Role::Conductor, WorkKind::Coordination)
                | (Role::Shepherd, Role::Coder, WorkKind::ProductionCode)
                | (Role::Shepherd, Role::Worker, WorkKind::Artifact)
                | (Role::Shepherd, Role::Auditor, WorkKind::Review)
                | (Role::Conductor, Role::Coder, WorkKind::ProductionCode)
                | (Role::Conductor, Role::Worker, WorkKind::Artifact)
                | (Role::Conductor, Role::Auditor, WorkKind::Review)
        ),
        Some(RunStatus::Closing) => matches!(
            (caller, target, work_kind),
            (Role::Shepherd, Role::Auditor, WorkKind::Review)
                | (Role::Shepherd, Role::Critic, WorkKind::Review)
                | (Role::Shepherd, Role::Discovery, WorkKind::Research)
        ),
        // `planned` is the gap between a verified plan and an opened sprint.
        // Nothing is dispatched inside it: planning work is finished and
        // execution has not been authorized. `closed` is terminal. Both were
        // already denied by the previous catch-all; they are written out so the
        // denial is a decision rather than a fallthrough.
        Some(RunStatus::Planned | RunStatus::Closed) | None => false,
    };
    if allowed {
        Ok(())
    } else {
        Err(DispatchError::PendingEdge(format!(
            "`{run_status}` does not authorize `{caller}` -> `{target}` for `{work_kind:?}`"
        )))
    }
}

fn validate_unique_scopes(scopes: &[PathAuthority], kind: &str) -> DispatchResult<()> {
    let mut seen = BTreeSet::new();
    for scope in scopes {
        if !seen.insert(scope.as_str()) {
            return Err(DispatchError::InvalidPending(format!(
                "duplicate {kind} scope `{scope}`"
            )));
        }
    }
    Ok(())
}

fn validate_attachment_identity(
    target: Harness,
    role: Role,
    agent_id: &AgentId,
    carrier_path: &str,
    startup_skill: &str,
    attachment_kind: AttachmentKind,
) -> DispatchResult<()> {
    if matches!(target, Harness::PrimeAgent) {
        return Err(DispatchError::AttachmentMismatch(
            "PrimeAgent has no pending carrier contract".into(),
        ));
    }
    attachment_kind.validate_for(target)?;
    let components = carrier_path.split('/').collect::<Vec<_>>();
    let component_start = usize::from(
        components.first() == Some(&"")
            || components.first().is_some_and(|part| {
                part.len() == 2
                    && part.as_bytes()[1] == b':'
                    && part.as_bytes()[0].is_ascii_alphabetic()
            }),
    );
    if !is_absolute_carrier_path(carrier_path)
        || carrier_path.is_empty()
        || carrier_path.len() > 4_096
        || !carrier_path.is_ascii()
        || carrier_path.contains(['\\', '\0'])
        || carrier_path.chars().any(char::is_control)
        || components[component_start..].iter().any(|part| {
            part.is_empty()
                || *part == "."
                || *part == ".."
                || part.ends_with('.')
                || part.ends_with(' ')
                || part.contains('~')
        })
    {
        return Err(DispatchError::AttachmentMismatch(
            "installed carrier path is not a bounded no-follow path".into(),
        ));
    }
    if startup_skill.is_empty()
        || startup_skill.len() > 64
        || !startup_skill
            .bytes()
            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-')
    {
        return Err(DispatchError::AttachmentMismatch(
            "startup skill identifier is invalid".into(),
        ));
    }
    if agent_id.as_str().is_empty() || role.is_root() {
        return Err(DispatchError::AttachmentMismatch(
            "carrier attachment must name a non-root child".into(),
        ));
    }
    Ok(())
}

fn is_absolute_carrier_path(value: &str) -> bool {
    value.starts_with('/')
        || (value.len() >= 3
            && value.as_bytes()[0].is_ascii_alphabetic()
            && value.as_bytes()[1] == b':'
            && value.as_bytes()[2] == b'/')
}

fn validate_digest(digest: [u8; 32], label: &str) -> DispatchResult<()> {
    if digest == [0; 32] {
        Err(DispatchError::InvalidPending(format!(
            "{label} digest must not be zero"
        )))
    } else {
        Ok(())
    }
}

impl Role {
    #[must_use]
    pub const fn is_root(self) -> bool {
        matches!(self, Self::Shepherd | Self::Planter)
    }

    #[must_use]
    pub const fn allows_work_kind(self, work_kind: WorkKind) -> bool {
        matches!(
            (self, work_kind),
            (Self::Engineer, WorkKind::Planning)
                | (Self::Coder, WorkKind::ProductionCode)
                | (Self::Worker, WorkKind::Artifact)
                | (Self::Auditor, WorkKind::Review)
                | (Self::Critic, WorkKind::Review)
                | (Self::Discovery, WorkKind::Research)
                | (Self::Conductor, WorkKind::Coordination)
        )
    }

    #[must_use]
    pub const fn write_eligible(self) -> bool {
        matches!(
            self,
            Self::Engineer
                | Self::Conductor
                | Self::Coder
                | Self::Worker
                | Self::Planter
                | Self::Shepherd
        )
    }
}