use std::{
collections::{BTreeMap, BTreeSet},
path::Path,
};
use sha2::{Digest, Sha256};
use shepherd::run::{LaneStatus, RunStatus};
use shepherd::{
dispatch::{
AgentId, DispatchBudget, DispatchBudgetInput, DispatchError, DispatchState, LaneId,
PendingDispatch, PendingLaunchState, ProjectId, Role,
},
plan::{LifecycleCapacity, PlanTopology, SessionReclamation, TurnStrategy},
};
use crate::{
ContextInputs, DispatchStore, DispatchStoreError, DispatchStoreResult, ExecutionContext,
dispatch_store::{DispatchInventory, LockedDispatchRun},
};
pub(crate) fn authorize(
workspace: &Path,
store: &DispatchStore,
project_id: &ProjectId,
pending: &PendingDispatch,
now: i64,
authority: &LockedDispatchRun<'_>,
) -> DispatchStoreResult<()> {
let mut inputs = ContextInputs::from_environment(workspace).map_err(budget_error)?;
inputs.active_harness = Some(pending.expected_attachment.target);
let context = ExecutionContext::discover(inputs).map_err(budget_error)?;
if context.workspace_root != workspace
|| context.runs_root != store.runs_root()
|| &pending.project_id != project_id
{
return Err(budget_error(
"dispatch budget workspace/project authority changed",
));
}
let state = store.load_run(&pending.run)?;
if state.status != pending.run_status {
return Err(budget_error(
"run status changed before dispatch budget authorization",
));
}
let plan = match state.status.known() {
Some(RunStatus::Executing) => Some(
crate::orientation::verified_execution_topology(&context, &state).map_err(|error| {
budget_error(error.message_text().unwrap_or("plan readiness is invalid"))
})?,
),
Some(RunStatus::Planted | RunStatus::Planned) => None,
Some(RunStatus::Closing | RunStatus::Closed) | None => {
return Err(budget_error("run is not open for bounded dispatch"));
}
};
let inventory = authority.inventory()?;
let current_root = store.load_current_root_binding(&pending.root_session_id)?;
if current_root.project_id != *project_id || current_root.run != pending.run {
return Err(budget_error(
"dispatch budget root session belongs to another project or run",
));
}
let wave_limit = plan
.as_ref()
.map(|plan| execution_admission(pending, &inventory, plan, &state))
.transpose()?
.flatten();
let budget = if wave_limit.is_some() {
measure_with_wave_limit(
pending,
&inventory,
context.config.spawn.max_parallel,
plan.as_ref(),
now,
wave_limit,
)?
} else {
measure_with_wave_limit(
pending,
&inventory,
context.config.spawn.max_parallel,
plan.as_ref(),
now,
None,
)?
};
budget.authorize_next()?;
tracing::debug!(run = %pending.run, live = budget.observed_live,
total = budget.observed_total, limit = budget.effective_live_limit,
"native dispatch budget authorized");
Ok(())
}
fn budget_error(error: impl core::fmt::Display) -> DispatchStoreError {
DispatchError::InvalidBudget(error.to_string()).into()
}
fn ceiling(value: usize) -> DispatchStoreResult<u32> {
u32::try_from(value).map_err(|_| budget_error("plan dispatch capacity exceeds u32"))
}
fn execution_admission(
candidate: &PendingDispatch,
inventory: &DispatchInventory,
plan: &PlanTopology,
state: &shepherd::RunState,
) -> DispatchStoreResult<Option<u32>> {
if state.run != plan.run || state.run != candidate.run.as_str() {
return Err(budget_error(
"native execution run and verified plan disagree",
));
}
let planned = plan
.lanes
.iter()
.map(|lane| lane.id.as_str())
.collect::<BTreeSet<_>>();
let mut lanes = BTreeMap::new();
for lane in &state.lanes {
if !planned.contains(lane.id.as_str())
|| lanes.insert(lane.id.as_str(), lane).is_some()
|| lane.plan != format!("lanes/{}/plan.md", lane.id)
|| !lane.state.is_known()
{
return Err(budget_error(
"native lane registration differs from the verified execution plan",
));
}
}
if lanes.len() != planned.len() {
return Err(budget_error("native lane registration is incomplete"));
}
let candidate_lane = candidate
.lane
.as_ref()
.ok_or_else(|| budget_error("execution dispatch requires a verified lane"))?;
let lane = lanes
.get(candidate_lane.as_str())
.ok_or_else(|| budget_error("dispatch lane is absent from the verified execution plan"))?;
if !matches!(
lane.state.known(),
Some(LaneStatus::Pending | LaneStatus::InProgress)
) {
return Err(budget_error(
"execution dispatch lane is already complete or in error",
));
}
if candidate.role == Role::Conductor {
let task = format!(".shepherd/runs/{}/lanes/{}/plan.md", state.run, lane.id);
let rendered = shepherd::plan::render_lane(plan, &lane.id).map_err(budget_error)?;
let digest: [u8; 32] = Sha256::digest(rendered).into();
if candidate.task_path.as_str() != task || candidate.task_sha256 != digest {
return Err(budget_error(
"Conductor task must be the exact rendered lane plan",
));
}
}
let (live, _) = inventory_agents(candidate, inventory)?;
let mut live_lanes = BTreeSet::from([candidate_lane.as_str()]);
for live_lane in live.values() {
let id = live_lane.as_ref().ok_or_else(|| {
budget_error("a lane-free planning process remains live during execution")
})?;
if !planned.contains(id.as_str()) {
return Err(budget_error(
"a live native process names an unplanned lane",
));
}
live_lanes.insert(id.as_str());
}
if live_lanes.len() > plan.capacity.logical_lane_limit {
return Err(budget_error(
"dispatch would exceed the verified logical lane limit",
));
}
if plan.capacity.schedule.is_empty() {
return Ok(None);
}
let wave = plan
.capacity
.schedule
.iter()
.find(|wave| {
wave.lanes.iter().any(|id| {
lanes
.get(id.as_str())
.is_some_and(|lane| !lane.state.is(LaneStatus::Complete))
})
})
.ok_or_else(|| budget_error("every verified capacity wave is complete"))?;
if !wave.lanes.iter().any(|id| id == candidate_lane.as_str()) {
return Err(budget_error(
"dispatch lane is outside the current capacity wave",
));
}
for live_lane in live.values() {
if !live_lane
.as_ref()
.is_some_and(|id| wave.lanes.iter().any(|lane| lane == id.as_str()))
{
return Err(budget_error(
"a live native process or reservation remains outside the current capacity wave",
));
}
}
Ok(Some(ceiling(wave.process_slots)?))
}
#[cfg(test)]
pub(crate) fn measure(
candidate: &PendingDispatch,
inventory: &DispatchInventory,
project_limit: u32,
plan: Option<&PlanTopology>,
now: i64,
) -> DispatchStoreResult<DispatchBudget> {
measure_with_wave_limit(candidate, inventory, project_limit, plan, now, None)
}
fn measure_with_wave_limit(
candidate: &PendingDispatch,
inventory: &DispatchInventory,
project_limit: u32,
plan: Option<&PlanTopology>,
now: i64,
wave_limit: Option<u32>,
) -> DispatchStoreResult<DispatchBudget> {
let harness = candidate.expected_attachment.target;
let limits = harness.limits();
let lifecycle = if let Some(plan) = plan {
Some(plan.capacity.lifecycle.as_ref().ok_or_else(|| {
budget_error("verified execution plan lacks lifecycle capacity evidence")
})?)
} else {
limits.lifecycle.as_ref()
};
let snapshot = capacity_agents(candidate, inventory, lifecycle)?;
let observed_live = ceiling(snapshot.occupied.len())?;
let observed_total = ceiling(snapshot.total.len())?;
let (host, plan_limit, parent_limit, run_limit) = if let Some(plan) = plan {
if plan.run != candidate.run.as_str()
|| !candidate
.lane
.as_ref()
.is_some_and(|lane| plan.lanes.iter().any(|planned| planned.id == lane.as_str()))
{
return Err(budget_error(
"dispatch lane is absent from the verified execution plan",
));
}
let capacity = &plan.capacity;
let lifecycle = lifecycle.expect("plan lifecycle was required above");
let host = ceiling(capacity.host_process_ceiling)?
.min(ceiling(lifecycle.live_concurrency_ceiling)?)
.min(ceiling(lifecycle.retained_descendant_slots)?);
let adapter_host = limits
.lifecycle
.as_ref()
.map(|lifecycle| ceiling(lifecycle.live_concurrency_ceiling))
.transpose()?
.or(limits.max_concurrent_agents);
(
Some(adapter_host.map_or(host, |limit| host.min(limit))),
ceiling(capacity.plan_process_ceiling)?.min(wave_limit.unwrap_or(u32::MAX)),
ceiling(capacity.parent_role_cap)?,
ceiling(capacity.run_budget)?,
)
} else {
(
limits
.lifecycle
.as_ref()
.map(|lifecycle| {
ceiling(
lifecycle
.live_concurrency_ceiling
.min(lifecycle.retained_descendant_slots),
)
})
.transpose()?
.or(limits.max_concurrent_agents),
project_limit,
project_limit,
project_limit,
)
};
let adapter_lifetime = limits
.lifecycle
.as_ref()
.and_then(|lifecycle| lifecycle.lifetime_descendant_slots)
.or_else(|| {
limits
.max_total_dispatches_per_run
.and_then(|limit| usize::try_from(limit).ok())
});
let plan_lifetime = plan
.and_then(|plan| plan.capacity.lifecycle.as_ref())
.and_then(|lifecycle| lifecycle.lifetime_descendant_slots);
let lifetime = match (adapter_lifetime, plan_lifetime) {
(Some(left), Some(right)) => Some(left.min(right)),
(Some(limit), None) | (None, Some(limit)) => Some(limit),
(None, None) => None,
};
let remaining = lifetime
.map(ceiling)
.transpose()?
.map(|limit| limit.saturating_sub(observed_total))
.map_or(run_limit, |remaining| remaining.min(run_limit));
let budget = DispatchBudget::measure(DispatchBudgetInput {
project_id: candidate.project_id.clone(),
run: candidate.run.clone(),
root_session_id: candidate.root_session_id.clone(),
parent_dispatch_id: candidate.parent_dispatch_id.clone(),
lane: candidate.lane.clone(),
parent_role: candidate.caller_role,
harness,
host_concurrent_limit: host,
project_parallel_limit: project_limit,
plan_parallel_limit: plan_limit,
parent_parallel_limit: parent_limit,
remaining_run_dispatches: remaining,
observed_live,
observed_total,
measured_at: now,
})?;
if !snapshot.retained_terminal.is_empty() {
match plan.and_then(|plan| plan.capacity.turn_strategy) {
Some(TurnStrategy::ResetBetweenPhases) => {
return Err(budget_error(format!(
"turn reset is not an authenticated Native capability; capacity consumers: {}; use a freshly bound root session or reuse the exact supported session",
snapshot.consumers.join("; ")
)));
}
Some(TurnStrategy::FreshRootSessionBetweenPhases) => {
return Err(budget_error(format!(
"fresh root session required before dispatch; capacity consumers: {}; bind the next host session to the exact run before continuing",
snapshot.consumers.join("; ")
)));
}
Some(TurnStrategy::SameTurn) | None => {}
}
}
if observed_live >= budget.effective_live_limit && !snapshot.terminal_consumers.is_empty() {
return Err(budget_error(format!(
"retained descendant capacity exhausted at {}/{} before provider launch; consumers: {}; bind a fresh root session, reuse an exact supported session, or reduce persistent lanes",
observed_live,
budget.effective_live_limit,
snapshot.consumers.join("; ")
)));
}
Ok(budget)
}
#[cfg(test)]
fn counts(
candidate: &PendingDispatch,
inventory: &DispatchInventory,
) -> DispatchStoreResult<(u32, u32)> {
let (live, total) = inventory_agents(candidate, inventory)?;
Ok((ceiling(live.len())?, ceiling(total.len())?))
}
type NativeAgents = (BTreeMap<AgentId, Option<LaneId>>, BTreeSet<AgentId>);
struct CapacitySnapshot {
occupied: BTreeMap<AgentId, Option<LaneId>>,
total: BTreeSet<AgentId>,
consumers: Vec<String>,
retained_terminal: Vec<String>,
terminal_consumers: Vec<String>,
}
fn capacity_agents(
candidate: &PendingDispatch,
inventory: &DispatchInventory,
lifecycle: Option<&LifecycleCapacity>,
) -> DispatchStoreResult<CapacitySnapshot> {
let (mut occupied, total) = inventory_agents(candidate, inventory)?;
let mut details = BTreeMap::<AgentId, String>::new();
for record in &inventory.records {
if record.state == DispatchState::Active && occupied.contains_key(&record.agent_id) {
details.insert(
record.agent_id.clone(),
format!(
"agent={} session={} role={} lane={} state={} turn={} reason=active",
record.agent_id,
record.session_id,
record.role,
record.lane.as_ref().map_or("<none>", LaneId::as_str),
record.state,
record
.observed_turn_id
.as_ref()
.map_or("<missing>", |turn| turn.as_str()),
),
);
}
}
for pending in &inventory.pending {
let agent = &pending.expected_attachment.agent_id;
if occupied.contains_key(agent) && !details.contains_key(agent) {
details.insert(
agent.clone(),
format!(
"agent={} session={} role={} lane={} state={} turn=<pending> reason=reservation",
agent,
pending.expected_child_session_id,
pending.role,
pending.lane.as_ref().map_or("<none>", LaneId::as_str),
pending.launch_state,
),
);
}
}
let mut retained_terminal = Vec::new();
let mut terminal_consumers = Vec::new();
if let Some(lifecycle) = lifecycle {
for record in &inventory.records {
if record.root_session_id != candidate.root_session_id || !record.state.is_terminal() {
continue;
}
let reclamation = if record.state == DispatchState::Stopped {
lifecycle.completed_session_reclamation
} else {
lifecycle.interrupted_session_reclamation
};
let reason = match reclamation {
SessionReclamation::Immediate => None,
SessionReclamation::Never => Some("never-reclaimed"),
SessionReclamation::TurnBoundary => Some("turn-boundary-untrusted"),
};
let Some(reason) = reason else {
continue;
};
occupied.insert(record.agent_id.clone(), record.lane.clone());
let detail = format!(
"agent={} session={} role={} lane={} state={} observed_turn={} reason={reason}",
record.agent_id,
record.session_id,
record.role,
record.lane.as_ref().map_or("<none>", LaneId::as_str),
record.state,
record
.observed_turn_id
.as_ref()
.map_or("<missing>", |turn| turn.as_str()),
);
retained_terminal.push(detail.clone());
terminal_consumers.push(detail.clone());
details.insert(record.agent_id.clone(), detail);
}
}
Ok(CapacitySnapshot {
occupied,
total,
consumers: details.into_values().collect(),
retained_terminal,
terminal_consumers,
})
}
fn inventory_agents(
candidate: &PendingDispatch,
inventory: &DispatchInventory,
) -> DispatchStoreResult<NativeAgents> {
let mut live = BTreeMap::<AgentId, Option<LaneId>>::new();
let mut total = BTreeSet::<AgentId>::new();
let mut records = BTreeMap::new();
for record in &inventory.records {
record.validate_loaded()?;
if record.project_id != candidate.project_id || record.run != candidate.run {
return Err(budget_error(
"foreign dispatch record in native budget inventory",
));
}
if records.insert(record.agent_id.clone(), record).is_some() {
return Err(budget_error(
"duplicate dispatch record in native budget inventory",
));
}
total.insert(record.agent_id.clone());
if record.root_session_id != candidate.root_session_id {
if record.state == DispatchState::Active {
return Err(budget_error(format!(
"active dispatch {} remains bound to prior root session {}; stop or recover it before opening a fresh capacity epoch",
record.agent_id, record.root_session_id
)));
}
continue;
}
if record.state == DispatchState::Active {
live.insert(record.agent_id.clone(), record.lane.clone());
}
}
let mut reservations = BTreeSet::new();
for pending in &inventory.pending {
pending.validate()?;
let agent = &pending.expected_attachment.agent_id;
if pending.project_id != candidate.project_id || pending.run != candidate.run {
return Err(budget_error(
"foreign pending launch in native budget inventory",
));
}
if !reservations.insert(agent.clone()) {
return Err(budget_error("multiple native reservations name one agent"));
}
if agent == &candidate.expected_attachment.agent_id {
if pending.launch_id_hash != candidate.launch_id_hash {
return Err(budget_error(
"another native reservation already owns this agent",
));
}
live.remove(agent);
total.remove(agent);
continue;
}
total.insert(agent.clone());
if pending.root_session_id != candidate.root_session_id {
if matches!(
pending.launch_state,
PendingLaunchState::Pending
| PendingLaunchState::ClaimedUnspawned
| PendingLaunchState::Active
) {
return Err(budget_error(format!(
"live reservation {} remains bound to prior root session {}; clean it before opening a fresh capacity epoch",
agent, pending.root_session_id
)));
}
continue;
}
if let Some(record) = records.get(agent) {
if record.session_id != pending.expected_child_session_id
|| record.root_session_id != pending.root_session_id
|| record.harness != pending.expected_attachment.target
|| record.role != pending.role
|| record.lane != pending.lane
{
return Err(budget_error(
"pending and active native identities disagree",
));
}
continue;
}
if matches!(
pending.launch_state,
PendingLaunchState::Pending
| PendingLaunchState::ClaimedUnspawned
| PendingLaunchState::Active
) {
live.insert(agent.clone(), pending.lane.clone());
}
}
Ok((live, total))
}
#[cfg(test)]
pub(crate) mod tests {
use super::*;
use shepherd::{
Harness,
dispatch::{
AgentType, AttachmentKind, CapabilityProbe, CarrierAttachmentExpectation,
DispatchRecord, DispatchStart, GitCommit, LaneId, PathAuthority, ProjectFilesystemId,
Role, RunId, SessionId, WorkKind,
},
plan::{
LifecycleCapacity, PlanCapacity, PlanLane, SessionReclamation, TurnResetBehavior,
TurnStrategy,
},
};
pub(crate) fn pending(marker: u8) -> PendingDispatch {
PendingDispatch {
schema: "shepherd.pending-dispatch/2".into(),
launch_id_hash: [marker; 32],
parent_process_hash: [2; 32],
project_id: ProjectId::new("018f47ce-72d7-7f64-9eb1-2f651d521c2a").unwrap(),
project_filesystem_id: ProjectFilesystemId::new("03".repeat(32)).unwrap(),
run: RunId::new("v657").unwrap(),
run_status: RunStatus::Executing.into(),
root_session_id: SessionId::new("root").unwrap(),
caller_role: Role::Shepherd,
parent_dispatch_id: None,
replaces_agent_id: None,
role: Role::Conductor,
work_kind: WorkKind::Coordination,
lane: Some(LaneId::new("lane-a").unwrap()),
baseline_commit: GitCommit::new("04".repeat(20)).unwrap(),
read_scope: vec![PathAuthority::new("docs/**").unwrap()],
write_scope: vec![],
result_artifact: PathAuthority::exact(
".shepherd/runs/v657/lanes/lane-a/reports/lead.md",
)
.unwrap(),
review_artifact: PathAuthority::exact(
".shepherd/runs/v657/lanes/lane-a/reviews/lead.md",
)
.unwrap(),
task_path: PathAuthority::exact("docs/task.md").unwrap(),
task_sha256: [5; 32],
expected_child_session_id: SessionId::new(format!("session-{marker}")).unwrap(),
expected_attachment: CarrierAttachmentExpectation {
target: Harness::Pi,
role: Role::Conductor,
agent_id: AgentId::new(format!("agent-{marker}")).unwrap(),
installed_carrier_path: "/native/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::PiSkillPath,
},
expires_at: 10_000,
launch_state: PendingLaunchState::Pending,
claimed_at: None,
child_process_hash: None,
activated_at: None,
nonce_sha256: [9; 32],
}
}
fn active(pending: &mut PendingDispatch) -> DispatchRecord {
pending.claim(100, [10; 32]).unwrap();
pending.activate(101, [10; 32]).unwrap();
let contract = Role::Conductor.dispatch_capability_contract().unwrap();
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".into(),
nonce: "nonce".into(),
harness: pending.expected_attachment.target,
agent_id: pending.expected_attachment.agent_id.clone(),
agent_type: AgentType::new(pending.role.as_str()).unwrap(),
role: pending.role,
lane: pending.lane.clone(),
parent_agent_id: None,
session_id: pending.expected_child_session_id.clone(),
observed_turn_id: None,
write_scope: vec![],
model: None,
capability_contract: contract,
capability_probe: CapabilityProbe::new(observed, "fixture", "native", None, 101)
.unwrap(),
startup_attachment: None,
attachment_nonce: None,
result_artifact: None,
result_nonce: None,
review_artifact: None,
review_nonce: None,
started_at: 101,
lease_expires_at: 10_000,
resumes_agent_id: None,
})
.unwrap()
}
#[test]
fn pi_adapter_capacity_reserves_every_pending_process_against_authenticated_live_ceiling() {
let next = pending(4);
let mut inventory = DispatchInventory {
records: vec![],
pending: vec![pending(1), pending(2)],
};
let budget = measure(&next, &inventory, 3, None, 1_000).unwrap();
assert_eq!(budget.host_concurrent_limit, Some(3));
assert_eq!(budget.effective_live_limit, 3);
budget.authorize_next().unwrap();
inventory.pending.push(pending(3));
assert!(matches!(
measure(&next, &inventory, 3, None, 1_000)
.unwrap()
.authorize_next(),
Err(DispatchError::HarnessLimit {
limit: 3,
observed: 4,
..
})
));
}
#[test]
fn claim_and_activation_replace_their_own_single_reservation() {
let next = pending(1);
let inventory = DispatchInventory {
records: vec![],
pending: vec![next.clone()],
};
assert_eq!(counts(&next, &inventory).unwrap(), (0, 0));
measure(&next, &inventory, 1, None, 1_000)
.unwrap()
.authorize_next()
.unwrap();
let mut collision = next.clone();
collision.launch_id_hash = [11; 32];
assert!(
counts(
&next,
&DispatchInventory {
records: vec![],
pending: vec![collision]
}
)
.is_err()
);
}
#[test]
fn terminal_record_releases_live_capacity_but_retains_lifetime_history() {
let mut reserved = pending(1);
let mut record = active(&mut reserved);
let inventory = DispatchInventory {
records: vec![record.clone()],
pending: vec![reserved.clone()],
};
assert_eq!(counts(&pending(2), &inventory).unwrap(), (1, 1));
assert_eq!(
measure(&pending(2), &inventory, 2, None, 20_000)
.unwrap()
.observed_live,
1
);
record
.stop(shepherd::dispatch::StopRequest {
agent_id: record.agent_id.clone(),
expected_revision: record.revision,
stopped_at: 200,
result_artifact: None,
observed_turn_id: None,
})
.unwrap();
assert_eq!(
counts(
&pending(2),
&DispatchInventory {
records: vec![record],
pending: vec![reserved]
}
)
.unwrap(),
(0, 1)
);
}
#[test]
fn capacity_topology_codex_requires_a_fresh_bound_root_for_retained_sessions() {
fn codex_pending(marker: u8, role: Role) -> PendingDispatch {
let mut value = pending(marker);
value.role = role;
value.expected_attachment.target = Harness::Codex;
value.expected_attachment.role = role;
value.expected_attachment.attachment_kind = AttachmentKind::CodexCustomAgent;
value
}
fn active_role(marker: u8, role: Role, root: &str) -> DispatchRecord {
let contract = role.dispatch_capability_contract().unwrap();
let observed = contract
.required
.union(&contract.optional)
.cloned()
.collect::<BTreeSet<_>>();
DispatchRecord::start(DispatchStart {
project_id: ProjectId::new("018f47ce-72d7-7f64-9eb1-2f651d521c2a").unwrap(),
run: RunId::new("v657").unwrap(),
root_session_id: SessionId::new(root).unwrap(),
run_incarnation: "incarnation".into(),
nonce: format!("nonce-{marker}"),
harness: Harness::Codex,
agent_id: AgentId::new(format!("agent-{marker}")).unwrap(),
agent_type: AgentType::new(role.as_str()).unwrap(),
role,
lane: (role != Role::Critic).then(|| LaneId::new("lane-a").unwrap()),
parent_agent_id: None,
session_id: SessionId::new(format!("session-{marker}")).unwrap(),
observed_turn_id: None,
write_scope: vec![],
model: None,
capability_contract: contract,
capability_probe: CapabilityProbe::new(observed, "fixture", "native", None, 101)
.unwrap(),
startup_attachment: None,
attachment_nonce: None,
result_artifact: None,
result_nonce: None,
review_artifact: None,
review_nonce: None,
started_at: 101,
lease_expires_at: 10_000,
resumes_agent_id: None,
})
.unwrap()
}
fn stop_in_turn(
mut record: DispatchRecord,
turn: &shepherd::dispatch::TurnId,
) -> DispatchRecord {
record
.stop(shepherd::dispatch::StopRequest {
agent_id: record.agent_id.clone(),
expected_revision: record.revision,
stopped_at: 200,
result_artifact: None,
observed_turn_id: Some(turn.clone()),
})
.unwrap();
record
}
let turn_1 = shepherd::dispatch::TurnId::new("turn-1").unwrap();
let forged_turn = shepherd::dispatch::TurnId::new("forged-turn-2").unwrap();
let conductor_a = active_role(1, Role::Conductor, "root");
let conductor_b = active_role(2, Role::Conductor, "root");
let critic = stop_in_turn(active_role(3, Role::Critic, "root"), &turn_1);
let mut plan = plan();
plan.capacity.logical_lane_limit = 2;
plan.capacity.host_process_ceiling = 3;
plan.capacity.project_spawn_max_parallel = 3;
plan.capacity.plan_process_ceiling = 3;
plan.capacity.parent_role_cap = 3;
plan.capacity.run_budget = 3;
plan.capacity.simultaneous_process_ceiling = 3;
plan.capacity.lifecycle = Harness::Codex.limits().lifecycle;
plan.capacity.turn_strategy = Some(TurnStrategy::FreshRootSessionBetweenPhases);
let worker = codex_pending(4, Role::Worker);
let inventory = DispatchInventory {
records: vec![conductor_a.clone(), conductor_b.clone(), critic],
pending: vec![],
};
let retained = measure(&worker, &inventory, 3, Some(&plan), 300)
.expect_err("the current root cannot release retained sessions from raw turn data");
let diagnostic = retained.to_string();
for marker in [
"fresh root session required",
"agent=agent-1",
"agent=agent-2",
"role=critic",
"session=session-3",
"observed_turn=turn-1",
] {
assert!(
diagnostic.contains(marker),
"missing `{marker}`: {diagnostic}"
);
}
let forged_inventory = DispatchInventory {
records: inventory
.records
.iter()
.cloned()
.map(|mut record| {
if record.state.is_terminal() {
record.observed_turn_id = Some(forged_turn.clone());
}
record
})
.collect(),
pending: vec![],
};
assert!(
measure(&worker, &forged_inventory, 3, Some(&plan), 301)
.unwrap_err()
.to_string()
.contains("fresh root session required"),
"caller-supplied turn correlation must never reclaim capacity"
);
let mut premature = codex_pending(4, Role::Worker);
premature.root_session_id = SessionId::new("fresh-root").unwrap();
let active_prior_root = measure(&premature, &inventory, 3, Some(&plan), 350)
.expect_err("a fresh binding cannot abandon active prior-root descendants");
assert!(
active_prior_root
.to_string()
.contains("active dispatch agent-1 remains bound to prior root session"),
"{active_prior_root}"
);
let terminal_a = stop_in_turn(conductor_a, &turn_1);
let terminal_b = stop_in_turn(conductor_b, &turn_1);
let mut next_worker = codex_pending(4, Role::Worker);
next_worker.root_session_id = SessionId::new("fresh-root").unwrap();
let fresh_conductor = active_role(5, Role::Conductor, "fresh-root");
let fresh_inventory = DispatchInventory {
records: vec![
terminal_a,
terminal_b,
inventory.records[2].clone(),
fresh_conductor,
],
pending: vec![],
};
measure(&next_worker, &fresh_inventory, 3, Some(&plan), 400)
.unwrap()
.authorize_next()
.expect("a freshly bound root starts a new three-slot capacity epoch");
}
#[test]
fn failed_and_canceled_launches_are_not_live_but_do_not_erase_run_usage() {
let mut canceled = pending(1);
canceled.cancel().unwrap();
let mut failed = pending(2);
failed.fail().unwrap();
assert_eq!(
counts(
&pending(3),
&DispatchInventory {
records: vec![],
pending: vec![canceled, failed]
}
)
.unwrap(),
(0, 2)
);
}
#[test]
fn malformed_or_cross_identity_inventory_never_becomes_spare_capacity() {
let next = pending(2);
let mut other = pending(1);
other.run = RunId::new("v656").unwrap();
assert!(
counts(
&next,
&DispatchInventory {
records: vec![],
pending: vec![other]
}
)
.is_err()
);
let other = pending(1);
assert!(
counts(
&next,
&DispatchInventory {
records: vec![],
pending: vec![other.clone(), other]
}
)
.is_err()
);
let mut other = pending(1);
let mut record = active(&mut other);
record.session_id = SessionId::new("forged-session").unwrap();
assert!(
counts(
&next,
&DispatchInventory {
records: vec![record],
pending: vec![other]
}
)
.is_err()
);
}
fn plan() -> PlanTopology {
PlanTopology {
schema: "shepherd.plan-topology/2".into(),
run: "v657".into(),
seed: "seed".into(),
mesh: "mesh".into(),
planning_evidence: "phase0".into(),
goal: "verified fixture".into(),
deliverables: vec![],
lanes: vec![PlanLane {
id: "lane-a".into(),
conductor: "conductor".into(),
cargo_target: "lane-a".into(),
node_ids: vec![],
deliverables: vec![],
}],
nodes: vec![],
topological_order: vec![],
capacity_policy: "queue".into(),
capacity: PlanCapacity {
logical_lane_limit: 1,
host_process_ceiling: 3,
project_spawn_max_parallel: 4,
plan_process_ceiling: 3,
parent_role_cap: 2,
run_budget: 3,
simultaneous_process_ceiling: 2,
per_lane_child_wave_ceiling: 8,
disk_min_mib: 1024,
model_quota: 3,
lifecycle: Some(
LifecycleCapacity {
live_concurrency_ceiling: 3,
retained_descendant_slots: 3,
lifetime_descendant_slots: None,
completed_session_reclamation: SessionReclamation::Immediate,
interrupted_session_reclamation: SessionReclamation::Immediate,
turn_reset_behavior: TurnResetBehavior::ReclaimsTerminal,
reusable_sessions: true,
nested_dispatch: true,
persistent_agent_cost: 1,
independent_reviewer_reachable: true,
capability_source: String::new(),
capability_evidence_sha256: String::new(),
}
.with_evidence("test-adapter"),
),
turn_strategy: Some(TurnStrategy::SameTurn),
backpressure: "queue".into(),
cargo_targets: vec![],
conductors: vec![],
schedule: vec![],
scale_outcome: None,
},
}
}
#[test]
fn verified_plan_parent_and_run_limits_can_only_tighten_the_project_cap() {
let mut plan = plan();
let next = pending(1);
let inventory = DispatchInventory::default();
assert_eq!(
measure(&next, &inventory, 4, Some(&plan), 1_000)
.unwrap()
.effective_live_limit,
2
);
plan.capacity.run_budget = 1;
assert_eq!(
measure(&next, &inventory, 4, Some(&plan), 1_000)
.unwrap()
.effective_live_limit,
1
);
plan.capacity.run_budget = 3;
assert_eq!(
measure(&next, &inventory, 1, Some(&plan), 1_000)
.unwrap()
.effective_live_limit,
1
);
plan.lanes[0].id = "not-our-lane".into();
assert!(measure(&next, &inventory, 4, Some(&plan), 1_000).is_err());
}
#[test]
fn zero_capacity_and_exhausted_host_lifetime_quota_deny_before_launch() {
use sha2::{Digest, Sha256};
let mut next = pending(255);
assert!(measure(&next, &DispatchInventory::default(), 0, None, 1_000).is_err());
next.expected_attachment.target = Harness::ClaudeCode;
next.expected_attachment.attachment_kind = AttachmentKind::ClaudePreload;
let limit = Harness::ClaudeCode
.limits()
.max_total_dispatches_per_run
.expect("Claude lifetime limit");
let mut inventory = DispatchInventory::default();
for index in 0..limit {
let mut old = pending(1);
old.expected_attachment.agent_id = AgentId::new(format!("archived-{index}")).unwrap();
old.expected_child_session_id =
SessionId::new(format!("archived-session-{index}")).unwrap();
old.launch_id_hash = Sha256::digest(index.to_be_bytes()).into();
old.cancel().unwrap();
inventory.pending.push(old);
}
assert_eq!(counts(&next, &inventory).unwrap(), (0, limit));
let error = measure(&next, &inventory, 4, None, 1_000).unwrap_err();
assert!(
error.to_string().contains("positive native facts"),
"{error}"
);
}
fn scheduled() -> (PlanTopology, shepherd::RunState, PendingDispatch) {
use sha2::{Digest, Sha256};
let mut plan = plan();
let mut second = plan.lanes[0].clone();
second.id = "lane-b".into();
second.conductor = "second-conductor".into();
second.cargo_target = "lane-b".into();
plan.lanes.push(second);
plan.capacity.schedule = vec![
shepherd::plan::CapacityWave {
lanes: vec!["lane-a".into()],
process_slots: 2,
},
shepherd::plan::CapacityWave {
lanes: vec!["lane-b".into()],
process_slots: 1,
},
];
let state = serde_json::from_value(serde_json::json!({
"run": "v657", "status": "executing", "lanes": [
{ "id": "lane-a", "plan": "lanes/lane-a/plan.md", "state": "pending" },
{ "id": "lane-b", "plan": "lanes/lane-b/plan.md", "state": "pending" }
]
}))
.unwrap();
let mut next = pending(2);
next.task_path = PathAuthority::exact(".shepherd/runs/v657/lanes/lane-a/plan.md").unwrap();
next.task_sha256 =
Sha256::digest(shepherd::plan::render_lane(&plan, "lane-a").unwrap()).into();
(plan, state, next)
}
#[test]
fn execution_schedule_reopens_only_the_first_incomplete_wave_with_its_own_ceiling() {
use sha2::{Digest, Sha256};
let (plan, mut state, mut next) = scheduled();
let empty = DispatchInventory::default();
assert_eq!(
execution_admission(&next, &empty, &plan, &state).unwrap(),
Some(2)
);
next.lane = Some(LaneId::new("lane-b").unwrap());
next.task_path = PathAuthority::exact(".shepherd/runs/v657/lanes/lane-b/plan.md").unwrap();
next.task_sha256 =
Sha256::digest(shepherd::plan::render_lane(&plan, "lane-b").unwrap()).into();
assert!(
execution_admission(&next, &empty, &plan, &state)
.unwrap_err()
.to_string()
.contains("current capacity wave")
);
state.lanes[0].state = LaneStatus::Complete.into();
assert_eq!(
execution_admission(&next, &empty, &plan, &state).unwrap(),
Some(1)
);
state.lanes[1].state = LaneStatus::Complete.into();
assert!(execution_admission(&next, &empty, &plan, &state).is_err());
}
#[test]
fn conductor_must_receive_the_exact_compiler_owned_lane_plan_not_an_arbitrary_brief() {
let (plan, state, mut next) = scheduled();
let empty = DispatchInventory::default();
next.task_path = PathAuthority::exact("docs/task.md").unwrap();
assert!(
execution_admission(&next, &empty, &plan, &state)
.unwrap_err()
.to_string()
.contains("exact rendered lane plan")
);
next.task_path = PathAuthority::exact(".shepherd/runs/v657/lanes/lane-a/plan.md").unwrap();
next.task_sha256 = [0; 32];
assert!(execution_admission(&next, &empty, &plan, &state).is_err());
next.role = Role::Coder;
assert!(
execution_admission(&next, &empty, &plan, &state).is_ok(),
"ordinary child briefs are not Conductor lane plans"
);
}
#[test]
fn scheduling_denies_missing_drifted_error_or_completed_native_lane_rows() {
let (plan, state, next) = scheduled();
let empty = DispatchInventory::default();
let mut changed = state.clone();
changed.lanes.pop();
assert!(execution_admission(&next, &empty, &plan, &changed).is_err());
let mut changed = state.clone();
changed.lanes[0].plan = "unreviewed/plan.md".into();
assert!(execution_admission(&next, &empty, &plan, &changed).is_err());
for terminal in ["error", "complete", "unknown"] {
let mut changed = state.clone();
changed.lanes[0].state = shepherd::run::Vocabulary::parse(terminal);
assert!(
execution_admission(&next, &empty, &plan, &changed).is_err(),
"{terminal}"
);
}
let mut changed = state.clone();
changed.lanes.push(state.lanes[0].clone());
assert!(execution_admission(&next, &empty, &plan, &changed).is_err());
}
#[test]
fn a_completed_wave_with_a_live_provider_or_pending_reservation_cannot_advance() {
use sha2::{Digest, Sha256};
let (plan, mut state, mut next) = scheduled();
state.lanes[0].state = LaneStatus::Complete.into();
next.lane = Some(LaneId::new("lane-b").unwrap());
next.task_path = PathAuthority::exact(".shepherd/runs/v657/lanes/lane-b/plan.md").unwrap();
next.task_sha256 =
Sha256::digest(shepherd::plan::render_lane(&plan, "lane-b").unwrap()).into();
let mut prior = pending(1);
let mut inventory = DispatchInventory {
records: vec![],
pending: vec![prior.clone()],
};
assert!(execution_admission(&next, &inventory, &plan, &state).is_err());
let mut record = active(&mut prior);
inventory = DispatchInventory {
records: vec![record.clone()],
pending: vec![prior.clone()],
};
assert!(execution_admission(&next, &inventory, &plan, &state).is_err());
record
.stop(shepherd::dispatch::StopRequest {
agent_id: record.agent_id.clone(),
expected_revision: record.revision,
stopped_at: 200,
result_artifact: None,
observed_turn_id: None,
})
.unwrap();
inventory.records = vec![record];
assert_eq!(
execution_admission(&next, &inventory, &plan, &state).unwrap(),
Some(1)
);
}
#[test]
fn unscheduled_execution_still_obeys_logical_lane_width_and_rejects_live_planning_leaks() {
let (mut plan, state, mut next) = scheduled();
plan.capacity.schedule.clear();
next.role = Role::Coder;
next.lane = Some(LaneId::new("lane-b").unwrap());
let mut other = pending(1);
let mut inventory = DispatchInventory {
records: vec![],
pending: vec![other.clone()],
};
assert!(
execution_admission(&next, &inventory, &plan, &state)
.unwrap_err()
.to_string()
.contains("logical lane limit")
);
plan.capacity.logical_lane_limit = 2;
assert!(execution_admission(&next, &inventory, &plan, &state).is_ok());
other.role = Role::Engineer;
other.work_kind = WorkKind::Planning;
other.lane = None;
other.expected_attachment.role = Role::Engineer;
inventory.pending = vec![other];
assert!(execution_admission(&next, &inventory, &plan, &state).is_err());
}
#[test]
fn a_capacity_wave_can_only_tighten_the_numeric_plan_limit() {
let (plan, _, next) = scheduled();
let inventory = DispatchInventory {
records: vec![],
pending: vec![pending(1)],
};
let budget =
measure_with_wave_limit(&next, &inventory, 4, Some(&plan), 100, Some(1)).unwrap();
assert_eq!(budget.effective_live_limit, 1);
assert!(budget.authorize_next().is_err());
let budget =
measure_with_wave_limit(&next, &inventory, 4, Some(&plan), 100, Some(100)).unwrap();
assert_eq!(budget.effective_live_limit, 2);
}
}