use std::{
collections::BTreeSet,
fs,
ops::Deref,
path::{Path, PathBuf},
process::{Command, Stdio},
sync::{Arc, Barrier},
thread,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use shepherd_cli::shepherd::run::RunStatus;
use shepherd_cli::shepherd::{
Harness, RunState,
dispatch::{
AgentId, AgentType, AttachmentKind, CapabilityProbe, CarrierAttachmentExpectation,
DispatchId, DispatchRecord, DispatchStart, DispatchState, GitCommit, LaneId,
NativeIdentity, PathAuthority, PendingDispatch, PendingLaunchState, ProjectFilesystemId,
ProjectId, ROOT_SESSION_SCHEMA, ReviewCustody, ReviewCustodyState, ReviewRuling,
ReviewVerdict, Role, RootSessionBinding, RunId, SessionId, StartupAttachment, StopRequest,
},
};
use shepherd_cli::{DispatchStore, DispatchStoreError, ReviewQuarantineFault, RunStore};
struct FixtureDir {
_guard: tempfile::TempDir,
root: PathBuf,
}
impl Deref for FixtureDir {
type Target = Path;
fn deref(&self) -> &Self::Target {
&self.root
}
}
impl AsRef<Path> for FixtureDir {
fn as_ref(&self) -> &Path {
&self.root
}
}
fn fixture(label: &str) -> FixtureDir {
let guard = tempfile::Builder::new()
.prefix(&format!("shepherd-dispatch-store-{label}-"))
.tempdir()
.expect("create fixture root");
let root = fs::canonicalize(guard.path()).expect("canonical fixture");
FixtureDir {
_guard: guard,
root,
}
}
fn write_run(runs: &Path, status: &str) {
write_run_for(runs, "v645", status);
}
fn write_run_for(runs: &Path, run: &str, status: &str) {
let state: RunState = serde_json::from_value(serde_json::json!({
"run": run,
"status": status,
}))
.expect("run state");
state
.store(&runs.join(run).join("run.json"))
.expect("run state");
}
fn pending(run: &RunId) -> PendingDispatch {
PendingDispatch {
schema: "shepherd.pending-dispatch/2".into(),
launch_id_hash: [1; 32],
parent_process_hash: [2; 32],
project_id: ProjectId::new("018f47ce-72d7-7f64-9eb1-2f651d521c2a").expect("project"),
project_filesystem_id: ProjectFilesystemId::new("03".repeat(32)).expect("filesystem"),
run: run.clone(),
run_status: RunStatus::Executing.into(),
root_session_id: SessionId::new("root-session").expect("root"),
caller_role: Role::Shepherd,
parent_dispatch_id: None,
replaces_agent_id: None,
role: Role::Conductor,
work_kind: shepherd_cli::shepherd::dispatch::WorkKind::Coordination,
lane: None,
baseline_commit: GitCommit::new("04".repeat(20)).expect("commit"),
read_scope: vec![PathAuthority::new("docs/**").expect("scope")],
write_scope: vec![],
result_artifact: PathAuthority::exact(".shepherd/runs/v645/reports/conductor.md")
.expect("result"),
review_artifact: PathAuthority::exact(".shepherd/runs/v645/reviews/conductor.md")
.expect("review"),
task_path: PathAuthority::exact("docs/task.md").expect("task"),
task_sha256: [5; 32],
expected_child_session_id: SessionId::new("child-session").expect("child"),
expected_attachment: CarrierAttachmentExpectation {
target: Harness::Pi,
role: Role::Conductor,
agent_id: shepherd_cli::shepherd::dispatch::AgentId::new("conductor-agent")
.expect("agent"),
installed_carrier_path: "/private/tmp/carrier.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: 200,
launch_state: PendingLaunchState::Pending,
claimed_at: None,
child_process_hash: None,
activated_at: None,
nonce_sha256: [9; 32],
}
}
fn binding() -> RootSessionBinding {
RootSessionBinding {
schema: ROOT_SESSION_SCHEMA.into(),
project_id: ProjectId::new("018f47ce-72d7-7f64-9eb1-2f651d521c2a").expect("project"),
run: RunId::new("v645").expect("run"),
harness: Harness::Pi,
session_id: SessionId::new("root-session").expect("session"),
role: Role::Shepherd,
mode: shepherd_cli::shepherd::dispatch::RootMode::Execution,
project_filesystem_id: None,
bound_at: 1_000,
expires_at: 2_000,
}
}
fn active_subject(run: &RunId) -> (PendingDispatch, DispatchRecord) {
let mut pending = pending(run);
pending.lane = Some(LaneId::new("lane-a").expect("lane"));
pending.expires_at = 10_000;
pending.claimed_at = Some(100);
pending.child_process_hash = Some([10; 32]);
pending.activated_at = Some(101);
pending.launch_state = PendingLaunchState::Active;
pending.validate().expect("active pending");
let contract = Role::Conductor
.dispatch_capability_contract()
.expect("Conductor contract");
let observed: BTreeSet<String> = contract
.required
.union(&contract.optional)
.cloned()
.collect();
let record = DispatchRecord::start(DispatchStart {
project_id: pending.project_id.clone(),
run: run.clone(),
root_session_id: pending.root_session_id.clone(),
run_incarnation: "incarnation-v645".into(),
nonce: "subject-nonce".into(),
harness: Harness::Pi,
agent_id: pending.expected_attachment.agent_id.clone(),
agent_type: AgentType::new("conductor").expect("agent type"),
role: Role::Conductor,
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", "fixture", None, 101)
.expect("probe"),
startup_attachment: None,
attachment_nonce: None,
result_artifact: Some(pending.result_artifact.as_str().into()),
result_nonce: Some("ab".repeat(32)),
review_artifact: None,
review_nonce: None,
started_at: 101,
lease_expires_at: pending.expires_at,
resumes_agent_id: None,
})
.expect("active subject record");
(pending, record)
}
fn terminal_worker_fixture(label: &str) -> (FixtureDir, DispatchStore, RunId, DispatchRecord) {
let root = fixture(label);
write_run(&root, "executing");
fs::create_dir_all(root.join("v645/dispatch")).expect("dispatch directory");
fs::create_dir_all(root.join("v645/lanes/lane-a/workers")).expect("worker directory");
fs::write(root.join("v645/lanes/lane-a/workers/evidence.txt"), b"ok").expect("task evidence");
fs::create_dir_all(root.join("v645/lanes/lane-a/reviews")).expect("review directory");
let run = RunId::new("v645").expect("run");
let mut pending = pending(&run);
pending.role = Role::Worker;
pending.work_kind = shepherd_cli::shepherd::dispatch::WorkKind::Artifact;
pending.lane = Some(LaneId::new("lane-a").expect("lane"));
pending.result_artifact =
PathAuthority::exact(".shepherd/runs/v645/lanes/lane-a/workers/worker.json")
.expect("worker result path");
pending.review_artifact =
PathAuthority::exact(".shepherd/runs/v645/lanes/lane-a/reviews/worker.json")
.expect("worker review path");
pending.write_scope = vec![pending.result_artifact.clone()];
pending.expected_attachment.role = Role::Worker;
pending.expected_attachment.agent_id = AgentId::new("worker-agent").expect("worker");
pending.expected_attachment.startup_skill = "artifact-work".into();
pending.claimed_at = Some(100);
pending.child_process_hash = Some([10; 32]);
pending.activated_at = Some(101);
pending.expires_at = 10_000;
pending.launch_state = PendingLaunchState::Active;
pending.validate().expect("worker pending");
let contract = Role::Worker
.dispatch_capability_contract()
.expect("Worker contract");
let observed = contract
.required
.union(&contract.optional)
.cloned()
.collect::<BTreeSet<_>>();
let record = DispatchRecord::start(DispatchStart {
project_id: pending.project_id.clone(),
run: run.clone(),
root_session_id: pending.root_session_id.clone(),
run_incarnation: "incarnation-v645".into(),
nonce: "aa".repeat(32),
harness: Harness::Pi,
agent_id: pending.expected_attachment.agent_id.clone(),
agent_type: AgentType::new("worker").expect("worker type"),
role: Role::Worker,
lane: pending.lane.clone(),
parent_agent_id: None,
session_id: pending.expected_child_session_id.clone(),
observed_turn_id: None,
write_scope: vec![pending.result_artifact.as_str().into()],
model: None,
capability_contract: contract,
capability_probe: CapabilityProbe::new(observed, "fixture", "fixture", None, 101)
.expect("probe"),
startup_attachment: Some(StartupAttachment {
skill: "artifact-work".into(),
bundle_digest: "08".repeat(32),
}),
attachment_nonce: Some("09".repeat(32)),
result_artifact: Some(pending.result_artifact.as_str().into()),
result_nonce: Some("0a".repeat(32)),
review_artifact: None,
review_nonce: None,
started_at: 101,
lease_expires_at: pending.expires_at,
resumes_agent_id: None,
})
.expect("worker record");
let store = DispatchStore::new(&root);
store
.publish_pending(&pending)
.expect("pending publication");
fs::write(
root.join("v645/dispatch/worker-agent.json"),
serde_json::to_vec(&record).expect("record JSON"),
)
.expect("record publication");
(root, store, run, record)
}
fn valid_worker_document(record: &DispatchRecord, pending: &PendingDispatch) -> serde_json::Value {
serde_json::json!({
"schema": "shepherd.worker-result/1",
"run": record.run,
"lane": record.lane,
"node": "worker-node",
"role": "worker",
"work_kind": "artifact",
"deliverable": "one evidence-bearing worker result",
"source_paths": ["docs/task.md"],
"owned_scope": [pending.result_artifact],
"budget": {"tool_calls": 2, "seconds": 30},
"output_shape": "typed JSON result with evidence",
"result_artifact": record.result_artifact,
"evidence": [{
"path": ".shepherd/runs/v645/lanes/lane-a/workers/evidence.txt",
"sha256": "2689367b205c16ce32ed4200942b8b8b1e262dfc70d9bc9fbc77c49699a4f1df",
"bytes": 2,
"command": "printf worker",
"exit_status": 0
}],
"status": "complete",
"task_digest": "05".repeat(32),
"startup_skill": "artifact-work",
"skill_bundle_digest": "08".repeat(32)
})
}
fn worker_native(record: &DispatchRecord) -> NativeIdentity {
NativeIdentity {
harness: record.harness,
project_id: record.project_id.clone(),
run: record.run.clone(),
lane: record.lane.clone(),
session_id: record.session_id.clone(),
agent_id: Some(record.agent_id.clone()),
agent_type: Some(record.agent_type.clone()),
role: Some(record.role),
tool_call_id: None,
now: 500,
root_binding: None,
}
}
fn add_auditor(
root: &Path,
store: &DispatchStore,
subject: &DispatchRecord,
subject_pending: &PendingDispatch,
) -> (PendingDispatch, DispatchRecord) {
let run = subject.run.clone();
let mut pending = subject_pending.clone();
pending.launch_id_hash = [11; 32];
pending.parent_process_hash = [13; 32];
pending.nonce_sha256 = [12; 32];
pending.child_process_hash = Some([14; 32]);
pending.role = Role::Auditor;
pending.work_kind = shepherd_cli::shepherd::dispatch::WorkKind::Review;
pending.expected_attachment.role = Role::Auditor;
pending.expected_attachment.agent_id = AgentId::new("auditor-agent").expect("auditor");
pending.expected_attachment.startup_skill = "reviewing".into();
pending.expected_child_session_id = SessionId::new("auditor-session").expect("session");
pending.result_artifact =
PathAuthority::exact(".shepherd/runs/v645/lanes/lane-a/reviews/auditor-result.json")
.expect("auditor result");
pending.review_artifact =
PathAuthority::exact(".shepherd/runs/v645/lanes/lane-a/reviews/auditor.json")
.expect("auditor review");
pending.claimed_at = Some(110);
pending.activated_at = Some(111);
pending.expires_at = 10_000;
pending.launch_state = PendingLaunchState::Active;
pending.validate().expect("auditor pending");
let contract = Role::Auditor
.dispatch_capability_contract()
.expect("Auditor contract");
let observed = contract
.required
.union(&contract.optional)
.cloned()
.collect::<BTreeSet<_>>();
let record = DispatchRecord::start(DispatchStart {
project_id: pending.project_id.clone(),
run,
root_session_id: pending.root_session_id.clone(),
run_incarnation: subject.run_incarnation.clone(),
nonce: "ab".repeat(32),
harness: Harness::Pi,
agent_id: pending.expected_attachment.agent_id.clone(),
agent_type: AgentType::new("auditor").expect("auditor type"),
role: Role::Auditor,
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", "fixture", None, 111)
.expect("probe"),
startup_attachment: Some(StartupAttachment {
skill: "reviewing".into(),
bundle_digest: "08".repeat(32),
}),
attachment_nonce: Some("0b".repeat(32)),
result_artifact: Some(pending.result_artifact.as_str().into()),
result_nonce: Some("0c".repeat(32)),
review_artifact: Some(pending.review_artifact.as_str().into()),
review_nonce: Some("0d".repeat(32)),
started_at: 111,
lease_expires_at: pending.expires_at,
resumes_agent_id: None,
})
.expect("auditor record");
store
.publish_pending(&pending)
.expect("auditor pending publication");
fs::write(
root.join("v645/dispatch/auditor-agent.json"),
serde_json::to_vec(&record).expect("auditor record JSON"),
)
.expect("auditor record publication");
(pending, record)
}
#[test]
fn native_terminal_boundary_rejects_empty_one_word_and_stale_worker_results() {
for (label, bytes) in [
("empty", Vec::new()),
("one-word", b"Done.".to_vec()),
(
"stale",
serde_json::to_vec(&serde_json::json!({
"schema": "shepherd.worker-result/1",
"run": "v645", "lane": "lane-a", "node": "n", "role": "worker",
"work_kind": "artifact", "deliverable": "stale", "source_paths": ["docs/task.md"],
"owned_scope": [".shepherd/runs/v645/lanes/lane-a/workers/worker.json"],
"budget": {"tool_calls": 1, "seconds": 1}, "output_shape": "stale",
"result_artifact": ".shepherd/runs/v645/lanes/lane-a/workers/worker.json",
"evidence": [{"path":".shepherd/runs/v645/lanes/lane-a/workers/evidence.txt","sha256":"01".repeat(32),"bytes":2,"command":"true","exit_status":0}],
"status": "complete", "task_digest": "ff".repeat(32),
"startup_skill": "artifact-work", "skill_bundle_digest": "08".repeat(32)
}))
.expect("stale result JSON"),
),
] {
let (root, store, run, record) = terminal_worker_fixture(&format!("terminal-{label}"));
fs::write(
root.join("v645/lanes/lane-a/workers/worker.json"),
bytes,
)
.expect("terminal artifact");
let error = store
.stop_verified_for_run(
&worker_native(&record),
StopRequest {
agent_id: record.agent_id.clone(),
expected_revision: record.revision,
stopped_at: 500,
result_artifact: record.result_artifact.clone(),
observed_turn_id: None,
},
)
.expect_err("invalid terminal result must be denied");
assert!(
error.to_string().contains("terminal") || error.to_string().contains("Worker"),
"unexpected {label} error: {error}"
);
assert_eq!(
store.load_for_run(&run, &record.agent_id).expect("record remains active").state,
DispatchState::Active
);
}
}
#[test]
fn native_terminal_boundary_accepts_typed_worker_and_stores_digest() {
let (root, store, run, record) = terminal_worker_fixture("terminal-valid");
let pending = store
.load_pending_for_agent(&run, &record.agent_id)
.expect("pending");
fs::write(
root.join("v645/lanes/lane-a/workers/worker.json"),
serde_json::to_vec(&valid_worker_document(&record, &pending)).expect("worker JSON"),
)
.expect("terminal artifact");
let stopped = store
.stop_verified_for_run(
&worker_native(&record),
StopRequest {
agent_id: record.agent_id.clone(),
expected_revision: record.revision,
stopped_at: 500,
result_artifact: record.result_artifact.clone(),
observed_turn_id: None,
},
)
.expect("typed worker result is accepted");
assert_eq!(stopped.state, DispatchState::Stopped);
assert!(stopped.result_sha256.is_some());
assert_eq!(
store
.load_for_run(&run, &record.agent_id)
.expect("stopped record")
.result_sha256,
stopped.result_sha256
);
fs::write(
root.join("v645/lanes/lane-a/workers/worker.json"),
b"post-terminal mutation",
)
.expect("mutate terminal artifact");
assert!(
store
.verify_terminal_digest(&run, &record.agent_id)
.is_err(),
"post-terminal result mutation must invalidate custody"
);
}
#[test]
fn native_reviewer_terminal_authenticates_both_artifacts_and_subject_bytes() {
let (root, store, run, subject) = terminal_worker_fixture("terminal-review");
let subject_pending = store
.load_pending_for_agent(&run, &subject.agent_id)
.expect("subject pending");
let subject_result = root.join("v645/lanes/lane-a/workers/worker.json");
let subject_doc = valid_worker_document(&subject, &subject_pending);
fs::write(
&subject_result,
serde_json::to_vec(&subject_doc).expect("subject JSON"),
)
.expect("subject result");
let subject = store
.stop_verified_for_run(
&worker_native(&subject),
StopRequest {
agent_id: subject.agent_id.clone(),
expected_revision: subject.revision,
stopped_at: 500,
result_artifact: subject.result_artifact.clone(),
observed_turn_id: None,
},
)
.expect("subject completion");
let (_auditor_pending, auditor) = add_auditor(&root, &store, &subject, &subject_pending);
let subject_digest = subject.result_sha256.clone().expect("subject digest");
let review_ref = auditor.review_artifact.clone().expect("review path");
let review = serde_json::json!({
"schema": "shepherd.review-result/1",
"run": "v645",
"lane": "lane-a",
"mode": "auditor-posthoc",
"reviewer_role": "auditor",
"candidate_commit": "0123456789abcdef0123456789abcdef01234567",
"input_digest": subject_digest,
"startup_skill": "reviewing",
"skill_bundle_digest": "08".repeat(32),
"result_channel": "native-result",
"review_artifact": review_ref,
"subject_result_artifact": subject.result_artifact,
"subject_task_digest": "05".repeat(32),
"verdict": "pass",
"findings": [],
"report_path": "reports/auditor.md"
});
let review_bytes = serde_json::to_vec(&review).expect("review JSON");
let result_path = root.join("v645/lanes/lane-a/reviews/auditor-result.json");
let review_path = root.join("v645/lanes/lane-a/reviews/auditor.json");
fs::write(&result_path, &review_bytes).expect("auditor result");
let mut mismatched = review.clone();
mismatched["input_digest"] = serde_json::json!("ff".repeat(32));
fs::write(
&review_path,
serde_json::to_vec(&mismatched).expect("mismatched review"),
)
.expect("mismatched review artifact");
let error = store
.stop_verified_for_run(
&NativeIdentity {
harness: auditor.harness,
project_id: auditor.project_id.clone(),
run: run.clone(),
lane: auditor.lane.clone(),
session_id: auditor.session_id.clone(),
agent_id: Some(auditor.agent_id.clone()),
agent_type: Some(auditor.agent_type.clone()),
role: Some(auditor.role),
tool_call_id: None,
now: 600,
root_binding: None,
},
StopRequest {
agent_id: auditor.agent_id.clone(),
expected_revision: auditor.revision,
stopped_at: 600,
result_artifact: auditor.result_artifact.clone(),
observed_turn_id: None,
},
)
.expect_err("mismatched reviewer artifacts must be rejected");
assert!(
error.to_string().contains("review"),
"unexpected error: {error}"
);
fs::write(&review_path, &review_bytes).expect("valid review artifact");
let stopped = store
.stop_verified_for_run(
&NativeIdentity {
harness: auditor.harness,
project_id: auditor.project_id.clone(),
run: run.clone(),
lane: auditor.lane.clone(),
session_id: auditor.session_id.clone(),
agent_id: Some(auditor.agent_id.clone()),
agent_type: Some(auditor.agent_type.clone()),
role: Some(auditor.role),
tool_call_id: None,
now: 601,
root_binding: None,
},
StopRequest {
agent_id: auditor.agent_id.clone(),
expected_revision: auditor.revision,
stopped_at: 601,
result_artifact: auditor.result_artifact.clone(),
observed_turn_id: None,
},
)
.expect("valid reviewer result");
assert!(stopped.review_sha256.is_some());
store
.verify_review_subject(&run, &auditor.agent_id)
.expect("subject is unchanged");
fs::write(&subject_result, b"post-review mutation").expect("mutate subject");
assert!(
store
.verify_review_subject(&run, &auditor.agent_id)
.is_err(),
"post-review subject mutation must be rejected"
);
fs::write(
&subject_result,
serde_json::to_vec(&subject_doc).expect("subject JSON"),
)
.expect("restore subject result");
fs::write(
root.join("v645/lanes/lane-a/workers/evidence.txt"),
b"evidence mutation",
)
.expect("mutate subject evidence");
assert!(
store
.verify_review_subject(&run, &auditor.agent_id)
.is_err(),
"mutating referenced evidence must invalidate review custody"
);
}
#[test]
fn dispatch_inventory_distinguishes_pending_reservations_from_child_records() {
let root = fixture("inventory-pending");
write_run(&root, "executing");
let store = DispatchStore::new(&root);
let run = RunId::new("v645").expect("run");
let (pending, record) = active_subject(&run);
store
.publish_pending(&pending)
.expect("publish reservation");
fs::write(
root.join("v645/dispatch/conductor-agent.json"),
serde_json::to_vec(&record).expect("record JSON"),
)
.expect("native record fixture");
assert_eq!(
store.list_for_run(&run).expect("child inventory"),
vec![record]
);
}
fn review_ruling(run: &RunId, subject: &AgentId, reviewer: u8, at: i64) -> ReviewRuling {
ReviewRuling {
schema: "shepherd.review-ruling/1".into(),
project_id: ProjectId::new("018f47ce-72d7-7f64-9eb1-2f651d521c2a").expect("project"),
run: run.clone(),
subject_agent_id: subject.clone(),
task_sha256: [5; 32],
task_generation: 1,
reviewer_dispatch_id: DispatchId::new(format!("auditor-{reviewer}")).expect("reviewer"),
findings_sha256: [reviewer; 32],
subject_result_sha256: Some([6; 32]),
review_sha256: Some([7; 32]),
verdict: ReviewVerdict::Redo,
ruled_at: at,
}
}
fn review_custodies(
run: &RunId,
pending: &PendingDispatch,
record: &DispatchRecord,
) -> (ReviewCustody, ReviewCustody) {
let first = review_ruling(run, &record.agent_id, 1, 201);
let mut custody = ReviewCustody::begin(
&first,
record.root_session_id.clone(),
record.session_id.clone(),
record.role,
record.lane.clone(),
pending.launch_id_hash,
)
.expect("review custody");
custody.apply(&first).expect("first ruling");
custody
.apply(&review_ruling(run, &record.agent_id, 2, 202))
.expect("second ruling");
custody
.apply(&review_ruling(run, &record.agent_id, 3, 203))
.expect("third ruling");
let active = custody.clone();
custody
.apply(&review_ruling(run, &record.agent_id, 4, 204))
.expect("fourth ruling");
assert_eq!(custody.state, ReviewCustodyState::Malignant);
(active, custody)
}
fn resume_request(record: &DispatchRecord) -> DispatchStart {
let contract = record
.role
.dispatch_capability_contract()
.expect("resume contract");
let observed: BTreeSet<String> = contract
.required
.union(&contract.optional)
.cloned()
.collect();
DispatchStart {
project_id: record.project_id.clone(),
run: record.run.clone(),
root_session_id: record.root_session_id.clone(),
run_incarnation: record.run_incarnation.clone(),
nonce: "replacement-nonce".into(),
harness: record.harness,
agent_id: AgentId::new("replacement-conductor").expect("replacement"),
agent_type: AgentType::new("conductor").expect("agent type"),
role: record.role,
lane: record.lane.clone(),
parent_agent_id: record.parent_agent_id.clone(),
session_id: SessionId::new("replacement-session").expect("session"),
observed_turn_id: None,
write_scope: record.write_scope.clone(),
model: None,
capability_contract: contract,
capability_probe: CapabilityProbe::new(observed, "fixture", "fixture", None, 300)
.expect("probe"),
startup_attachment: record.startup_attachment.clone(),
attachment_nonce: record.attachment_nonce.clone(),
result_artifact: record.result_artifact.clone(),
result_nonce: Some("cd".repeat(32)),
review_artifact: record.review_artifact.clone(),
review_nonce: record.review_nonce.clone(),
started_at: 300,
lease_expires_at: record.lease_expires_at,
resumes_agent_id: Some(record.agent_id.clone()),
}
}
#[test]
fn fourth_rejection_commit_recovers_both_crash_seams_and_revokes_every_child_path() {
for fault in [
ReviewQuarantineFault::AfterCustodyCommit,
ReviewQuarantineFault::AfterRecordCommit,
] {
let root = fixture(&format!("review-quarantine-{fault:?}"));
let runs = root.join("runs");
write_run(&runs, "executing");
let run = RunId::new("v645").expect("run");
let store = DispatchStore::new(&runs);
let (active_pending, active_record) = active_subject(&run);
store
.publish_pending(&active_pending)
.expect("active pending fixture");
fs::write(
runs.join("v645/dispatch/conductor-agent.json"),
serde_json::to_vec(&active_record).expect("record json"),
)
.expect("active record fixture");
let (active_custody, malignant_custody) =
review_custodies(&run, &active_pending, &active_record);
store
.publish_review_custody(&active_custody)
.expect("active review custody");
let fault_store = DispatchStore::new(&runs).with_review_quarantine_fault(fault);
let error = fault_store
.quarantine_malignant(
Some(&active_custody),
&active_record,
&active_pending,
&malignant_custody,
)
.expect_err("injected commit seam");
assert!(matches!(error, DispatchStoreError::Reconciliation(_)));
let custody: ReviewCustody = serde_json::from_slice(
&fs::read(runs.join("v645/dispatch/.review-custody.conductor-agent.json"))
.expect("committed custody"),
)
.expect("custody json");
assert_eq!(custody.state, ReviewCustodyState::Malignant);
let record = fault_store
.load_for_run(&run, &active_record.agent_id)
.expect("first authority load performs idempotent terminal recovery");
assert_eq!(record.state, DispatchState::Malignant);
let recovered = fault_store
.load_review_custody(&run, &active_record.agent_id)
.expect("recovered custody");
assert_eq!(recovered, malignant_custody);
let pending = fault_store
.load_pending(&run, active_pending.launch_id_hash)
.expect("quarantined pending");
assert_eq!(pending.launch_state, PendingLaunchState::Quarantined);
assert!(matches!(
record.resume(resume_request(&record)),
Err(shepherd_cli::shepherd::dispatch::DispatchError::ReviewCustodyTerminal)
));
assert!(
fault_store
.claim_pending_unspawned(
&run,
active_pending.launch_id_hash,
[42; 32],
|_, _, _| Ok(()),
)
.is_err(),
"a quarantined broker claim cannot be reclaimed"
);
let mut replay = active_pending.clone();
replay.launch_id_hash = [11; 32];
replay.launch_state = PendingLaunchState::Pending;
replay.claimed_at = None;
replay.child_process_hash = None;
replay.activated_at = None;
assert!(
fault_store.publish_pending(&replay).is_err(),
"a replacement launch cannot reuse the destroyed child identity"
);
}
}
#[test]
fn process_claim_helper() {
let Some(runs) = std::env::var_os("SHEPHERD_STORE_RACE_RUNS") else {
return;
};
let marker = PathBuf::from(std::env::var_os("SHEPHERD_STORE_RACE_MARKER").expect("marker"));
let byte = std::env::var("SHEPHERD_STORE_RACE_BYTE")
.expect("race byte")
.parse::<u8>()
.expect("race byte is numeric");
let run = RunId::new("v645").expect("run");
let success = DispatchStore::new(runs)
.claim_pending_unspawned(&run, [1; 32], [byte; 32], |_, _, _| Ok(()))
.is_ok();
fs::write(marker, if success { b"1" } else { b"0" }).expect("marker");
}
#[test]
fn dispatch_and_run_close_share_the_per_run_lock() {
let root = fixture("shared-lock");
let runs = root.join("runs");
write_run(&runs, "executing");
let store = DispatchStore::new(&runs);
store
.publish_root_binding(&binding())
.expect("root binding");
assert!(runs.join("v645/run.lock").is_file());
assert!(!runs.join("v645/dispatch/.dispatch.lock").exists());
}
#[test]
fn two_process_claims_have_one_winner() {
let root = fixture("process-claim-race");
let runs = root.join("runs");
write_run(&runs, "executing");
let store = DispatchStore::new(&runs);
let run = RunId::new("v645").expect("run");
let mut prepared = pending(&run);
prepared.expires_at = i64::try_from(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("clock")
.as_millis(),
)
.expect("millisecond clock fits i64")
+ 60_000;
store.publish_pending(&prepared).expect("publish pending");
let executable = std::env::current_exe().expect("test executable");
let mut children = Vec::new();
for byte in [10_u8, 11_u8] {
let marker = root.join(format!("winner-{byte}"));
children.push(
Command::new(&executable)
.args(["--exact", "process_claim_helper", "--nocapture"])
.env("SHEPHERD_STORE_RACE_RUNS", &runs)
.env("SHEPHERD_STORE_RACE_MARKER", &marker)
.env("SHEPHERD_STORE_RACE_BYTE", byte.to_string())
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.expect("claim process"),
);
}
for child in children {
assert!(
child
.wait_with_output()
.expect("claim process output")
.status
.success()
);
}
let wins = [10_u8, 11_u8]
.into_iter()
.filter(|byte| {
fs::read(root.join(format!("winner-{byte}"))).expect("winner marker") == b"1"
})
.count();
assert_eq!(wins, 1);
}
#[test]
fn dispatch_transition_waits_for_the_run_close_lock() {
let root = fixture("close-lock-contention");
let runs = root.join("runs");
write_run(&runs, "executing");
let run_path = runs.join("v645/run.json");
let barrier = Arc::new(Barrier::new(2));
let holder_barrier = Arc::clone(&barrier);
let holder = thread::spawn(move || {
RunStore::with_timeout(&run_path, Duration::from_secs(1))
.update(|_| {
holder_barrier.wait();
thread::sleep(Duration::from_millis(150));
Ok(())
})
.expect("run close lock holder");
});
barrier.wait();
let error = DispatchStore::with_timeout(&runs, Duration::from_millis(20))
.publish_root_binding(&binding())
.expect_err("dispatch transition must wait for the shared run lock");
assert!(matches!(error, DispatchStoreError::LockTimeout { .. }));
holder.join().expect("lock holder");
}
#[test]
fn missing_launch_identity_never_becomes_a_claim() {
let root = fixture("missing-launch");
let runs = root.join("runs");
write_run(&runs, "executing");
let store = DispatchStore::new(&runs);
let error = store
.load_pending(&RunId::new("v645").expect("run"), [1; 32])
.expect_err("unknown launch");
assert!(matches!(error, DispatchStoreError::PendingNotFound { .. }));
}
#[test]
fn restart_reconciles_claimed_unspawned_and_cleanup_is_one_shot() {
let root = fixture("reconcile");
let runs = root.join("runs");
write_run(&runs, "executing");
let store = DispatchStore::new(&runs);
let run = RunId::new("v645").expect("run");
let mut prepared = pending(&run);
prepared.validate().expect("pending fixture");
store.publish_pending(&prepared).expect("publish pending");
prepared.claimed_at = Some(100);
prepared.child_process_hash = Some([10; 32]);
prepared.launch_state = PendingLaunchState::ClaimedUnspawned;
prepared.validate().expect("claimed fixture");
let path = runs.join("v645/dispatch/pending-");
let hash: String = prepared
.launch_id_hash
.iter()
.map(|byte| format!("{byte:02x}"))
.collect();
let path = PathBuf::from(format!("{}{}.json", path.display(), hash));
fs::write(&path, serde_json::to_vec(&prepared).expect("pending json")).expect("claimed bytes");
assert_eq!(
store.reconcile_unspawned_at(&run, 101).expect("reconcile"),
1
);
let expired = store
.load_pending(&run, prepared.launch_id_hash)
.expect("expired pending");
assert_eq!(expired.launch_state, PendingLaunchState::Expired);
assert!(
store
.cleanup_pending(&run, prepared.launch_id_hash, PendingLaunchState::Canceled)
.is_err()
);
}
#[test]
fn pending_claim_reads_authority_under_the_existing_run_lock() {
let root = fixture("claim-authority-lock");
let runs = root.join("runs");
write_run(&runs, "executing");
let store = DispatchStore::with_timeout(&runs, Duration::from_millis(20));
let run = RunId::new("v645").expect("run");
let root_binding = binding();
store
.publish_root_binding(&root_binding)
.expect("root binding");
let (mut prepared, record) = active_subject(&run);
let (custody, _) = review_custodies(&run, &prepared, &record);
prepared.launch_state = PendingLaunchState::Pending;
prepared.claimed_at = None;
prepared.child_process_hash = None;
prepared.activated_at = None;
prepared.expires_at = i64::try_from(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("clock")
.as_millis(),
)
.expect("clock fits i64")
+ 60_000;
store.publish_pending(&prepared).expect("pending");
store
.publish_review_custody(&custody)
.expect("review custody");
let claimed = store
.claim_pending_unspawned(
&run,
prepared.launch_id_hash,
[42; 32],
|pending, _, authority| {
assert_eq!(
authority.load_root_binding(&pending.root_session_id)?,
root_binding
);
assert_eq!(
authority.load_review_custody(&pending.expected_attachment.agent_id)?,
custody
);
assert!(
matches!(
store.load_root_binding_for_run(&run, &root_binding.session_id),
Err(DispatchStoreError::LockTimeout { .. }),
),
"the view must not release the transition lock"
);
Ok(())
},
)
.expect("authority reads must not recursively acquire the run lock");
assert_eq!(claimed.launch_state, PendingLaunchState::ClaimedUnspawned);
}
#[test]
fn concurrent_claims_have_one_winner_and_no_active_record() {
let root = fixture("claim-race");
let runs = root.join("runs");
write_run(&runs, "executing");
let store = DispatchStore::new(&runs);
let run = RunId::new("v645").expect("run");
let mut prepared = pending(&run);
prepared.expires_at = i64::try_from(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("clock")
.as_millis(),
)
.expect("millisecond clock fits i64")
+ 60_000;
store.publish_pending(&prepared).expect("publish pending");
let store = Arc::new(store);
let mut workers = Vec::new();
for child in [[10; 32], [11; 32]] {
let store = Arc::clone(&store);
let worker_run = run.clone();
workers.push(thread::spawn(move || {
store
.claim_pending_unspawned(&worker_run, prepared.launch_id_hash, child, |_, _, _| {
Ok(())
})
.is_ok()
}));
}
let wins = workers
.into_iter()
.map(|worker| worker.join().expect("claim worker"))
.filter(|winner| *winner)
.count();
assert_eq!(wins, 1);
let claimed = store
.load_pending(&run, prepared.launch_id_hash)
.expect("claimed");
assert_eq!(claimed.launch_state, PendingLaunchState::ClaimedUnspawned);
assert!(!runs.join("v645/dispatch/conductor-agent.json").exists());
}
#[test]
fn recovery_terminalizes_in_flight_rows_without_promoting_orphan_records() {
let root = fixture("recovery-no-orphan");
let runs = root.join("runs");
write_run(&runs, "executing");
let store = DispatchStore::new(&runs);
let run = RunId::new("v645").expect("run");
let mut prepared = pending(&run);
prepared.expires_at = 10_000;
store.publish_pending(&prepared).expect("publish pending");
prepared.claimed_at = Some(100);
prepared.child_process_hash = Some([10; 32]);
prepared.activated_at = Some(200);
prepared.launch_state = PendingLaunchState::Active;
prepared.validate().expect("active fixture");
let hash: String = prepared
.launch_id_hash
.iter()
.map(|byte| format!("{byte:02x}"))
.collect();
let path = runs.join(format!("v645/dispatch/pending-{hash}.json"));
fs::write(&path, serde_json::to_vec(&prepared).expect("pending json"))
.expect("active pending bytes");
assert_eq!(
store.reconcile_unspawned_at(&run, 201).expect("reconcile"),
1
);
let recovered = store
.load_pending(&run, prepared.launch_id_hash)
.expect("recovered pending");
assert_eq!(recovered.launch_state, PendingLaunchState::LaunchFailed);
assert!(!runs.join("v645/dispatch/conductor-agent.json").exists());
}
#[test]
fn closed_runs_reject_new_root_bindings() {
let root = fixture("closed");
let runs = root.join("runs");
write_run(&runs, "closed");
let error = DispatchStore::new(&runs)
.publish_root_binding(&binding())
.expect_err("closed run");
assert!(!matches!(error, DispatchStoreError::AlreadyExists { .. }));
}
#[test]
fn root_mode_transition_is_atomic_and_preserves_the_trusted_principal() {
let root = fixture("root-mode-transition-principal");
let runs = root.join("runs");
write_run(&runs, "planned");
let store = DispatchStore::new(&runs);
let mut planning = binding();
planning.mode = shepherd_cli::shepherd::dispatch::RootMode::Planting;
store
.publish_root_binding_for_run(&planning)
.expect("planning binding");
write_run(&runs, "executing");
let mut forged = planning.clone();
forged.mode = shepherd_cli::shepherd::dispatch::RootMode::Execution;
forged.harness = Harness::Codex;
forged.bound_at = 1_001;
forged.expires_at = 2_001;
store
.transition_root_binding_to_execution(&planning, &forged)
.expect_err("a mode transition must not change the trusted harness");
assert_eq!(
store
.load_root_binding_for_run(&planning.run, &planning.session_id)
.expect("unchanged planning binding"),
planning
);
let mut execution = planning.clone();
execution.mode = shepherd_cli::shepherd::dispatch::RootMode::Execution;
execution.bound_at = 1_001;
execution.expires_at = 2_001;
store
.transition_root_binding_to_execution(&planning, &execution)
.expect("same trusted principal advances atomically");
assert_eq!(
store
.load_root_binding_for_run(&execution.run, &execution.session_id)
.expect("execution binding"),
execution
);
store
.transition_root_binding_to_execution(&planning, &execution)
.expect_err("a stale planning compare must not overwrite the current binding");
}
#[test]
fn current_root_binding_moves_once_across_runs_and_rejects_replay() {
let root = fixture("current-root-cross-run");
let runs = root.join("runs");
write_run_for(&runs, "v645", "executing");
write_run_for(&runs, "v657", "planted");
let store = DispatchStore::new(&runs);
let old = binding();
store
.publish_root_binding_for_run(&old)
.expect("old per-run binding");
store
.activate_current_root_binding(&old)
.expect("initial current binding");
let mut next = old.clone();
next.run = RunId::new("v657").expect("next run");
next.mode = shepherd_cli::shepherd::dispatch::RootMode::Planting;
next.bound_at += 1;
next.expires_at += 1;
store
.publish_root_binding_for_run(&next)
.expect("next per-run binding");
store
.activate_current_root_binding(&next)
.expect("same root continues to the planted run");
assert_eq!(
store
.load_current_root_binding(&next.session_id)
.expect("current root binding"),
next
);
assert_eq!(
store
.load_root_binding_for_run(&old.run, &old.session_id)
.expect("old binding remains immutable evidence"),
old
);
store
.activate_current_root_binding(&old)
.expect_err("an old-run root binding cannot regain authority by replay");
let mut wrong_carrier = next.clone();
wrong_carrier.run = RunId::new("v658").expect("other run");
wrong_carrier.harness = Harness::Codex;
wrong_carrier.bound_at += 1;
wrong_carrier.expires_at += 1;
store
.activate_current_root_binding(&wrong_carrier)
.expect_err("cross-carrier continuation is not the same trusted principal");
}