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::PlanTopology,
};
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 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(
pending,
&inventory,
context.config.spawn.max_parallel,
plan.as_ref(),
now,
)?
};
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)?))
}
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 (observed_live, observed_total) = counts(candidate, inventory)?;
let harness = candidate.expected_attachment.target;
let limits = harness.limits();
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 host = ceiling(capacity.host_process_ceiling)?;
(
Some(
limits
.max_concurrent_agents
.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.max_concurrent_agents,
project_limit,
project_limit,
project_limit,
)
};
let remaining = limits
.max_total_dispatches_per_run
.map(|limit| limit.saturating_sub(observed_total))
.map_or(run_limit, |remaining| remaining.min(run_limit));
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,
})
.map_err(Into::into)
}
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>);
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.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 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::{PlanCapacity, PlanLane},
};
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: Harness::Pi,
agent_id: pending.expected_attachment.agent_id.clone(),
agent_type: AgentType::new("conductor").unwrap(),
role: Role::Conductor,
lane: pending.lane.clone(),
parent_agent_id: None,
session_id: pending.expected_child_session_id.clone(),
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 unbounded_host_still_reserves_every_pending_process_against_project_cap() {
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, None);
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,
})
.unwrap();
assert_eq!(
counts(
&pending(2),
&DispatchInventory {
records: vec![record],
pending: vec![reserved]
}
)
.unwrap(),
(0, 1)
);
}
#[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,
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
.unwrap();
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,
})
.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);
}
}