#[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,
validate_parent_child, validate_write_scope_pattern,
};
pub const DISPATCH_SCHEMA: &str = "shepherd.dispatch/3";
#[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,
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,
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,
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>,
pub result_nonce: Option<String>,
pub review_artifact: 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>,
}
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,
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_nonce: input.result_nonce,
review_artifact: input.review_artifact,
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(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(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(),
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);
if self.result_artifact.is_none() {
self.result_artifact = request.result_artifact;
}
self.revision += 1;
Ok(())
}
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 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:?}"),
})
}
}