shepherd-core 6.7.1

//! Versioned dispatch record creation, stop, and cross-harness resume.

#[cfg(feature = "alloc")]
use alloc::{
    format,
    string::{String, ToString},
    vec::Vec,
};

use crate::Harness;

use super::{
    AgentId, AgentType, CapabilityContract, CapabilityProbe, CapabilityReadiness, CapabilityReport,
    DispatchError, DispatchResult, LaneId, ProjectId, Role, RunId, SessionId, StartupAttachment,
    TurnId, validate_parent_child, validate_write_scope_pattern,
};

pub const DISPATCH_SCHEMA: &str = "shepherd.dispatch/3";

// The six role result contracts. Each names the exact document a dispatched role
// returns, and each was named ONLY in doctrine prose
// (content/skills/*/references/*-contract.md) with nothing in Rust defining it.
// A schema identifier that exists in one language and not the other is how a
// producer and a verifier drift apart without either side failing -- the release
// receipt did exactly that, emitting `shepherd.publish/4` against a verifier
// expecting schema 3, after both registries had already been written.
//
// Binding the identifier here is the first half. The second half is deriving the
// SHAPE from a type rather than restating it in prose, and it lives in
// `result.rs`: one type per contract, each bound back to the constant below by
// an associated `SCHEMA`, each deriving `schemars::JsonSchema` behind the
// `schema` feature the way `ShepherdConfig` already does. Add an identifier here
// and it has no shape; add it there and it cannot get the wrong name.
pub const CODER_RESULT_SCHEMA: &str = "shepherd.coder-result/1";
pub const WORKER_RESULT_SCHEMA: &str = "shepherd.worker-result/1";
pub const DISCOVERY_REPORT_SCHEMA: &str = "shepherd.discovery-report/1";
pub const DEBUGGING_EVIDENCE_SCHEMA: &str = "shepherd.debugging-evidence/1";
pub const LANE_LEDGER_SCHEMA: &str = "shepherd.lane-ledger/1";

#[derive(
    Clone,
    Copy,
    Debug,
    Eq,
    Hash,
    Ord,
    PartialEq,
    PartialOrd,
    serde::Deserialize,
    serde::Serialize,
    strum::AsRefStr,
    strum::Display,
    strum::EnumCount,
    strum::EnumIs,
    strum::EnumString,
    strum::IntoStaticStr,
    strum::VariantNames,
)]
#[serde(rename_all = "snake_case")]
#[strum(ascii_case_insensitive, serialize_all = "snake_case")]
pub enum DispatchState {
    Active,
    CapabilityBlocked,
    Malignant,
    Stopped,
}

impl DispatchState {
    #[must_use]
    pub const fn is_terminal(self) -> bool {
        matches!(
            self,
            Self::CapabilityBlocked | Self::Malignant | Self::Stopped
        )
    }
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct DispatchStart {
    pub project_id: ProjectId,
    pub run: RunId,
    /// Root session whose current run binding owns this child capability.
    pub root_session_id: SessionId,
    pub run_incarnation: String,
    pub nonce: String,
    pub harness: Harness,
    pub agent_id: AgentId,
    pub agent_type: AgentType,
    pub role: Role,
    pub lane: Option<LaneId>,
    pub parent_agent_id: Option<AgentId>,
    pub session_id: SessionId,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub observed_turn_id: Option<TurnId>,
    pub write_scope: Vec<String>,
    pub model: Option<String>,
    pub capability_contract: CapabilityContract,
    pub capability_probe: CapabilityProbe,
    pub startup_attachment: Option<StartupAttachment>,
    pub attachment_nonce: Option<String>,
    pub result_artifact: Option<String>,
    pub result_nonce: Option<String>,
    pub review_artifact: Option<String>,
    pub review_nonce: Option<String>,
    pub started_at: i64,
    pub lease_expires_at: i64,
    pub resumes_agent_id: Option<AgentId>,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct DispatchRecord {
    pub schema: String,
    pub revision: u64,
    pub project_id: ProjectId,
    pub run: RunId,
    pub root_session_id: SessionId,
    pub run_incarnation: String,
    pub nonce: String,
    pub harness: Harness,
    pub agent_id: AgentId,
    pub agent_type: AgentType,
    pub role: Role,
    pub lane: Option<LaneId>,
    pub parent_agent_id: Option<AgentId>,
    pub session_id: SessionId,
    /// Adapter-reported turn associated with this lifecycle state. It is
    /// diagnostic correlation only and cannot release retained capacity.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub observed_turn_id: Option<TurnId>,
    pub write_scope: Vec<String>,
    pub model: Option<String>,
    pub capabilities: CapabilityReport,
    pub startup_attachment: Option<StartupAttachment>,
    pub attachment_nonce: Option<String>,
    pub state: DispatchState,
    pub started_at: i64,
    pub lease_expires_at: i64,
    pub stopped_at: Option<i64>,
    pub result_artifact: Option<String>,
    /// Digest captured by Native while the terminal transition holds the run
    /// lock. It is deliberately separate from the caller-provided path.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub result_sha256: Option<String>,
    pub result_nonce: Option<String>,
    pub review_artifact: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub review_sha256: Option<String>,
    pub review_nonce: Option<String>,
    pub resumes_agent_id: Option<AgentId>,
}

#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
#[serde(deny_unknown_fields)]
pub struct StopRequest {
    pub agent_id: AgentId,
    pub expected_revision: u64,
    pub stopped_at: i64,
    pub result_artifact: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub observed_turn_id: Option<TurnId>,
}

impl DispatchRecord {
    pub fn start(input: DispatchStart) -> DispatchResult<Self> {
        validate_start(&input)?;
        let compiled_contract = input.role.dispatch_capability_contract()?;
        if input.capability_contract != compiled_contract {
            return Err(DispatchError::InvalidRecord(
                "capability contract does not match the compiled role profile".into(),
            ));
        }
        let capabilities = input.capability_contract.evaluate(input.capability_probe);
        let blocked = capabilities.readiness() == CapabilityReadiness::Blocked;
        let record = Self {
            schema: DISPATCH_SCHEMA.into(),
            revision: 1,
            project_id: input.project_id,
            run: input.run,
            root_session_id: input.root_session_id,
            run_incarnation: input.run_incarnation,
            nonce: input.nonce,
            harness: input.harness,
            agent_id: input.agent_id,
            agent_type: input.agent_type,
            role: input.role,
            lane: input.lane,
            parent_agent_id: input.parent_agent_id,
            session_id: input.session_id,
            observed_turn_id: input.observed_turn_id,
            write_scope: input.write_scope,
            model: input.model,
            capabilities,
            startup_attachment: input.startup_attachment,
            attachment_nonce: input.attachment_nonce,
            state: if blocked {
                DispatchState::CapabilityBlocked
            } else {
                DispatchState::Active
            },
            started_at: input.started_at,
            lease_expires_at: input.lease_expires_at,
            stopped_at: blocked.then_some(input.started_at),
            result_artifact: input.result_artifact,
            result_sha256: None,
            result_nonce: input.result_nonce,
            review_artifact: input.review_artifact,
            review_sha256: None,
            review_nonce: input.review_nonce,
            resumes_agent_id: input.resumes_agent_id,
        };
        record.validate_loaded()?;
        Ok(record)
    }

    pub fn validate_loaded(&self) -> DispatchResult<()> {
        if self.schema != DISPATCH_SCHEMA {
            return Err(DispatchError::InvalidRecord(format!(
                "unsupported schema `{}`",
                self.schema
            )));
        }
        if self.revision == 0 {
            return Err(DispatchError::InvalidRecord(
                "revision must be positive".into(),
            ));
        }
        self.capabilities.validate()?;
        if let Some(reference) = &self.result_artifact {
            validate_artifact(reference)?;
        }
        if let Some(digest) = &self.result_sha256 {
            validate_digest_text(digest, "result digest")?;
            if self.result_artifact.is_none() {
                return Err(DispatchError::InvalidRecord(
                    "result digest has no native completion artifact".into(),
                ));
            }
        }
        if let Some(nonce) = &self.result_nonce {
            validate_nonce(nonce, "result")?;
            if self.result_artifact.is_none() {
                return Err(DispatchError::InvalidRecord(
                    "result nonce has no native completion artifact".into(),
                ));
            }
        }
        if let Some(reference) = &self.review_artifact {
            if !matches!(self.role, Role::Auditor | Role::Critic) {
                return Err(DispatchError::InvalidArtifact(reference.clone()));
            }
            validate_artifact(reference)?;
        }
        if let Some(digest) = &self.review_sha256 {
            validate_digest_text(digest, "review digest")?;
            if self.review_artifact.is_none() {
                return Err(DispatchError::InvalidRecord(
                    "review digest has no native review artifact".into(),
                ));
            }
        }
        if let Some(nonce) = &self.review_nonce {
            validate_nonce(nonce, "review")?;
            if self.review_artifact.is_none() {
                return Err(DispatchError::InvalidRecord(
                    "review nonce has no native review artifact".into(),
                ));
            }
        }
        if let Some(attachment) = &self.startup_attachment {
            attachment.validate()?;
            let nonce = self.attachment_nonce.as_deref().ok_or_else(|| {
                DispatchError::InvalidRecord("startup attachment nonce is missing".into())
            })?;
            if nonce.len() != 64
                || !nonce
                    .bytes()
                    .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
            {
                return Err(DispatchError::InvalidRecord(
                    "startup attachment nonce is invalid".into(),
                ));
            }
        } else if self.attachment_nonce.is_some() {
            return Err(DispatchError::InvalidRecord(
                "startup attachment nonce has no attachment".into(),
            ));
        }
        let expected_capabilities =
            self.role
                .dispatch_capability_contract()?
                .evaluate(CapabilityProbe {
                    probe_id: self.capabilities.probe_id.clone(),
                    observed: self.capabilities.observed.clone(),
                    observed_events: self.capabilities.observed_events.clone(),
                    source: self.capabilities.source.clone(),
                    harness_version: self.capabilities.harness_version.clone(),
                    provider_version: self.capabilities.provider_version.clone(),
                    binary_sha256: self.capabilities.binary_sha256.clone(),
                    package_sha256: self.capabilities.package_sha256.clone(),
                    probed_at: self.capabilities.probed_at,
                });
        if self.capabilities != expected_capabilities {
            return Err(DispatchError::InvalidRecord(
                "capability diff does not match the compiled role profile".into(),
            ));
        }
        let start = DispatchStart {
            project_id: self.project_id.clone(),
            run: self.run.clone(),
            root_session_id: self.root_session_id.clone(),
            run_incarnation: self.run_incarnation.clone(),
            nonce: self.nonce.clone(),
            harness: self.harness,
            agent_id: self.agent_id.clone(),
            agent_type: self.agent_type.clone(),
            role: self.role,
            lane: self.lane.clone(),
            parent_agent_id: self.parent_agent_id.clone(),
            session_id: self.session_id.clone(),
            observed_turn_id: self.observed_turn_id.clone(),
            write_scope: self.write_scope.clone(),
            model: self.model.clone(),
            capability_contract: CapabilityContract::default(),
            capability_probe: CapabilityProbe {
                probe_id: self.capabilities.probe_id.clone(),
                observed: self.capabilities.observed.clone(),
                observed_events: self.capabilities.observed_events.clone(),
                source: self.capabilities.source.clone(),
                harness_version: self.capabilities.harness_version.clone(),
                provider_version: self.capabilities.provider_version.clone(),
                binary_sha256: self.capabilities.binary_sha256.clone(),
                package_sha256: self.capabilities.package_sha256.clone(),
                probed_at: self.capabilities.probed_at,
            },
            started_at: self.started_at,
            lease_expires_at: self.lease_expires_at,
            startup_attachment: self.startup_attachment.clone(),
            attachment_nonce: self.attachment_nonce.clone(),
            result_artifact: self.result_artifact.clone(),
            result_nonce: self.result_nonce.clone(),
            review_artifact: self.review_artifact.clone(),
            review_nonce: self.review_nonce.clone(),
            resumes_agent_id: self.resumes_agent_id.clone(),
        };
        validate_start(&start)?;
        start.capability_probe.validate()?;
        match self.state {
            DispatchState::Active
                if self.revision == 1
                    && self.stopped_at.is_none()
                    && self.capabilities.readiness() != CapabilityReadiness::Blocked => {}
            DispatchState::CapabilityBlocked
                if self.revision == 1
                    && self.stopped_at == Some(self.started_at)
                    && self.capabilities.readiness() == CapabilityReadiness::Blocked => {}
            DispatchState::Malignant
                if self.revision >= 2
                    && self.stopped_at.is_some_and(|at| at >= self.started_at) => {}
            DispatchState::Stopped
                if self.revision >= 2
                    && self.stopped_at.is_some_and(|at| at >= self.started_at) => {}
            _ => {
                return Err(DispatchError::InvalidRecord(
                    "state, revision, timestamps, and capability readiness disagree".into(),
                ));
            }
        }
        Ok(())
    }

    pub fn stop(&mut self, request: StopRequest) -> DispatchResult<()> {
        if request.expected_revision != self.revision {
            return Err(DispatchError::RevisionMismatch {
                expected: request.expected_revision,
                found: self.revision,
            });
        }
        if request.agent_id != self.agent_id {
            return Err(DispatchError::AgentMismatch {
                expected: self.agent_id.to_string(),
                found: request.agent_id.to_string(),
            });
        }
        if self.state != DispatchState::Active {
            return Err(DispatchError::InvalidTransition {
                from: self.state,
                to: DispatchState::Stopped,
            });
        }
        if request.stopped_at < self.started_at {
            return Err(DispatchError::InvalidTime(
                "stop time precedes start time".into(),
            ));
        }
        if let Some(reference) = &request.result_artifact {
            validate_artifact(reference)?;
            if self
                .result_artifact
                .as_ref()
                .is_some_and(|expected| expected != reference)
            {
                return Err(DispatchError::InvalidArtifact(reference.clone()));
            }
        }
        self.state = DispatchState::Stopped;
        self.stopped_at = Some(request.stopped_at);
        self.observed_turn_id = request.observed_turn_id;
        if self.result_artifact.is_none() {
            self.result_artifact = request.result_artifact;
        }
        self.revision += 1;
        Ok(())
    }

    /// Attach the exact bytes observed by Native at the terminal boundary.
    /// Callers cannot supply this through `StopRequest`; the store computes it
    /// from the allocated artifact while holding the run lock.
    pub fn bind_terminal_digests(
        &mut self,
        result_sha256: String,
        review_sha256: Option<String>,
    ) -> DispatchResult<()> {
        if self.state != DispatchState::Stopped {
            return Err(DispatchError::InvalidTransition {
                from: self.state,
                to: DispatchState::Stopped,
            });
        }
        validate_digest_text(&result_sha256, "result digest")?;
        if review_sha256.is_some() && self.review_artifact.is_none() {
            return Err(DispatchError::InvalidRecord(
                "review digest has no native review artifact".into(),
            ));
        }
        if let Some(digest) = &review_sha256 {
            validate_digest_text(digest, "review digest")?;
        }
        self.result_sha256 = Some(result_sha256);
        self.review_sha256 = review_sha256;
        self.validate_loaded()
    }

    pub fn quarantine_malignant(&mut self, at: i64) -> DispatchResult<()> {
        if self.state != DispatchState::Active || at < self.started_at {
            return Err(DispatchError::ReviewCustodyTerminal);
        }
        self.state = DispatchState::Malignant;
        self.stopped_at = Some(at);
        self.revision = self
            .revision
            .checked_add(1)
            .ok_or_else(|| DispatchError::InvalidRecord("revision overflow".into()))?;
        self.validate_loaded()
    }

    pub fn resume(&self, input: DispatchStart) -> DispatchResult<Self> {
        if self.state == DispatchState::Malignant {
            return Err(DispatchError::ReviewCustodyTerminal);
        }
        if !self.state.is_terminal() {
            return Err(DispatchError::InvalidTransition {
                from: self.state,
                to: DispatchState::Active,
            });
        }
        if input.agent_id == self.agent_id {
            return Err(DispatchError::ReusedResumeIdentity);
        }
        if input.resumes_agent_id.as_ref() != Some(&self.agent_id) {
            return Err(DispatchError::ResumeMismatch {
                field: "resumes_agent_id",
                expected: self.agent_id.to_string(),
                found: input
                    .resumes_agent_id
                    .as_ref()
                    .map(ToString::to_string)
                    .unwrap_or_default(),
            });
        }
        require_resume_match("project_id", &self.project_id, &input.project_id)?;
        require_resume_match("run", &self.run, &input.run)?;
        require_resume_match(
            "root_session_id",
            &self.root_session_id,
            &input.root_session_id,
        )?;
        require_resume_match(
            "run_incarnation",
            &self.run_incarnation,
            &input.run_incarnation,
        )?;
        require_resume_match("role", &self.role, &input.role)?;
        require_resume_match("lane", &self.lane, &input.lane)?;
        require_resume_match(
            "parent_agent_id",
            &self.parent_agent_id,
            &input.parent_agent_id,
        )?;
        if self.write_scope != input.write_scope {
            return Err(DispatchError::ResumeMismatch {
                field: "write_scope",
                expected: format!("{:?}", self.write_scope),
                found: format!("{:?}", input.write_scope),
            });
        }
        Self::start(input)
    }
}

fn validate_start(input: &DispatchStart) -> DispatchResult<()> {
    input.capability_contract.validate()?;
    validate_token("run incarnation", &input.run_incarnation)?;
    validate_token("dispatch nonce", &input.nonce)?;
    input.capability_probe.validate()?;
    if input.harness == Harness::ClaudeCode
        && input.agent_type.as_str() != input.role.as_str()
        && input.agent_type.as_str() != input.role.carrier()
    {
        return Err(DispatchError::AgentTypeRoleMismatch {
            agent_type: input.agent_type.to_string(),
            role: input.role,
        });
    }
    if input.started_at < 0 || input.lease_expires_at <= input.started_at {
        return Err(DispatchError::InvalidTime(
            "lease must expire after a non-negative start".into(),
        ));
    }
    if input.role == Role::Conductor && input.lane.is_none() {
        return Err(DispatchError::InvalidRecord(
            "Conductor dispatches require a lane".into(),
        ));
    }
    validate_parent_child(input.parent_agent_id.as_ref(), &input.agent_id)?;
    if input.resumes_agent_id.as_ref() == Some(&input.agent_id) {
        return Err(DispatchError::ReusedResumeIdentity);
    }
    let mut scopes = input.write_scope.clone();
    scopes.sort();
    if scopes.windows(2).any(|pair| pair[0] == pair[1]) {
        return Err(DispatchError::InvalidWriteScope(
            "write_scope entries must be unique".into(),
        ));
    }
    for scope in &input.write_scope {
        validate_write_scope_pattern(scope)?;
    }
    if input.model.as_ref().is_some_and(|model| {
        model.is_empty() || model.len() > 256 || model.chars().any(char::is_control)
    }) {
        return Err(DispatchError::InvalidIdentifier {
            kind: "model",
            value: input.model.clone().unwrap_or_default(),
        });
    }
    Ok(())
}

fn validate_token(kind: &'static str, value: &str) -> DispatchResult<()> {
    if value.is_empty() || value.len() > 128 || value.chars().any(char::is_control) {
        return Err(DispatchError::InvalidIdentifier {
            kind,
            value: value.into(),
        });
    }
    Ok(())
}

fn validate_artifact(reference: &str) -> DispatchResult<()> {
    let valid = !reference.is_empty()
        && reference.len() <= 512
        && !reference.starts_with('/')
        && !reference.contains('\\')
        && !reference.contains('\0')
        && !reference.chars().any(char::is_control)
        && reference
            .split('/')
            .all(|part| !part.is_empty() && part != "." && part != "..");
    if valid {
        Ok(())
    } else {
        Err(DispatchError::InvalidArtifact(reference.into()))
    }
}

fn validate_nonce(value: &str, kind: &str) -> DispatchResult<()> {
    if value.len() != 64
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
    {
        return Err(DispatchError::InvalidRecord(format!(
            "{kind} nonce is not lowercase hexadecimal"
        )));
    }
    Ok(())
}

fn validate_digest_text(value: &str, kind: &str) -> DispatchResult<()> {
    if value.len() != 64
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
    {
        return Err(DispatchError::InvalidRecord(format!(
            "{kind} is not lowercase hexadecimal"
        )));
    }
    Ok(())
}

fn require_resume_match<T>(field: &'static str, expected: &T, found: &T) -> DispatchResult<()>
where
    T: Eq + core::fmt::Debug,
{
    if expected == found {
        Ok(())
    } else {
        Err(DispatchError::ResumeMismatch {
            field,
            expected: format!("{expected:?}"),
            found: format!("{found:?}"),
        })
    }
}