#[cfg(unix)]
#[path = "support/plan.rs"]
mod plan_support;
#[cfg(unix)]
mod unix {
use std::{
fs,
os::unix::fs::{FileTypeExt, MetadataExt, PermissionsExt},
path::{Path, PathBuf},
process::{Command, Stdio},
thread,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use sha2::{Digest, Sha256};
use shepherd_cli::shepherd::{
Harness, RunState,
compiler::{HarnessProfile, compile},
dispatch::{
AgentId, AttachmentKind, DispatchId, LoadedCarrierAttestationV1, PendingDispatch,
ProjectId, ReviewCustody, ReviewCustodyState, ReviewRuling, ReviewVerdict, Role, RunId,
},
registry::{Registry, SingletonPublicationState},
};
use shepherd_cli::{
BrokerError, CarrierAttachmentExpectationRequest, DispatchService, DispatchStore,
NativeBroker, PreparePendingDispatchRequest, ProcessIdentity,
};
fn fixture(label: &str) -> PathBuf {
let suffix = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("clock")
.as_nanos();
let path = std::env::temp_dir().join(format!("shepherd-broker-{label}-{suffix:x}"));
fs::create_dir_all(&path).expect("fixture");
fs::set_permissions(&path, fs::Permissions::from_mode(0o700)).expect("private fixture");
fs::canonicalize(path).expect("canonical fixture")
}
fn broker_endpoint(label: &str) -> PathBuf {
let nonce = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("clock")
.as_nanos();
fs::canonicalize("/tmp")
.expect("short Unix temporary root")
.join(format!("sb-{}-{nonce:x}-{label}", std::process::id()))
.join("broker.sock")
}
fn hex(bytes: &[u8; 32]) -> String {
bytes.iter().map(|byte| format!("{byte:02x}")).collect()
}
fn carrier_digest(path: &std::path::Path, bytes: &[u8]) -> [u8; 32] {
let mut content = Sha256::new();
content.update(b"regular-file\0");
content.update(0o100644_u32.to_be_bytes());
content.update((bytes.len() as i64).to_be_bytes());
content.update(bytes);
let content = content.finalize();
let metadata = fs::metadata(path).expect("carrier metadata");
let mut identity = Sha256::new();
identity.update(b"unix-identity\0");
identity.update(metadata.dev().to_be_bytes());
identity.update(metadata.ino().to_be_bytes());
identity.update(metadata.nlink().to_be_bytes());
identity.update(metadata.mode().to_be_bytes());
identity.update(metadata.len().to_be_bytes());
let identity = identity.finalize();
let mut digest = Sha256::new();
digest.update(b"carrier-identity/1\0");
digest.update(content);
digest.update(identity);
digest.finalize().into()
}
fn digest_from_hex(value: &str) -> [u8; 32] {
assert_eq!(value.len(), 64, "SHA-256 fixture length");
let mut output = [0_u8; 32];
for (index, pair) in value.as_bytes().as_chunks::<2>().0.iter().enumerate() {
let digit = |byte: u8| match byte {
b'0'..=b'9' => byte - b'0',
b'a'..=b'f' => byte - b'a' + 10,
_ => panic!("digest fixture"),
};
output[index] = digit(pair[0]) << 4 | digit(pair[1]);
}
output
}
fn wait_for_file(path: &std::path::Path) {
let deadline = std::time::Instant::now() + Duration::from_secs(10);
while !path.is_file() {
assert!(
std::time::Instant::now() < deadline,
"timed out at {}",
path.display()
);
thread::sleep(Duration::from_millis(5));
}
}
fn wait_for_claim(path: &std::path::Path, error: &std::path::Path) {
let deadline = std::time::Instant::now() + Duration::from_secs(10);
while !path.is_file() {
if error.is_file() {
panic!(
"child failed: {}",
fs::read_to_string(error).unwrap_or_default()
);
}
assert!(
std::time::Instant::now() < deadline,
"timed out at {}",
path.display()
);
thread::sleep(Duration::from_millis(5));
}
}
fn wait_for_pending_state(path: &Path, expected: &str) -> serde_json::Value {
let deadline = std::time::Instant::now() + Duration::from_secs(10);
loop {
let value: serde_json::Value =
serde_json::from_slice(&fs::read(path).expect("pending bytes"))
.expect("pending JSON");
if value["launch_state"] == expected {
return value;
}
assert!(
std::time::Instant::now() < deadline,
"pending state never reached {expected}: {value}"
);
thread::sleep(Duration::from_millis(5));
}
}
fn loaded_attestation(root: &Path, nonce_sha256: [u8; 32]) -> LoadedCarrierAttestationV1 {
let input = shepherd_cli::content_compiler::embedded_compile_input()
.expect("canonical embedded content");
let tree = compile(&input, &HarnessProfile::pi()).expect("canonical Pi carrier");
let manifest: serde_json::Value = serde_json::from_slice(
&fs::read(root.join(".shepherd-generated.json")).expect("installed manifest"),
)
.expect("manifest JSON");
assert_eq!(manifest["schema"], "shepherd.compiled-tree/4");
assert_eq!(manifest["target"], "pi");
assert_eq!(manifest["tree_digest"], tree.digest);
assert_eq!(
manifest["files"].as_array().expect("inventory").len(),
tree.files.len()
);
for file in &tree.files {
let installed = root.join(&file.path);
assert_eq!(
fs::read(&installed).expect("installed file"),
file.content.as_bytes()
);
assert_eq!(
fs::metadata(&installed).expect("installed mode").mode() & 0o777,
file.mode
);
}
let role = manifest["roles"]
.as_array()
.expect("roles")
.iter()
.find(|role| role["role"] == "conductor")
.expect("compiled Conductor role");
let compiled_role = tree
.roles
.iter()
.find(|role| role.role == "conductor")
.expect("canonical Conductor role");
assert_eq!(
role["startup_skill_sha256"],
compiled_role
.startup_skill_sha256
.as_deref()
.expect("bundle digest")
);
let carrier = root.join(role["carrier_path"].as_str().expect("carrier path"));
let carrier_bytes = fs::read(&carrier).expect("loaded Conductor carrier");
let candidate = fs::read(std::env::current_exe().expect("native test candidate"))
.expect("candidate bytes");
LoadedCarrierAttestationV1 {
schema: "shepherd.loaded-carrier/1".into(),
nonce_sha256,
target: Harness::Pi,
role: Role::Conductor,
agent_id: AgentId::new("conductor-agent").expect("agent"),
installed_carrier_path: carrier.display().to_string(),
candidate_sha256: Sha256::digest(candidate).into(),
carrier_sha256: carrier_digest(&carrier, &carrier_bytes),
compiler_tree_sha256: digest_from_hex(
manifest["tree_digest"].as_str().expect("tree digest"),
),
startup_skill: role["startup_skill"]
.as_str()
.expect("startup skill")
.into(),
skill_bundle_sha256: digest_from_hex(
role["startup_skill_sha256"]
.as_str()
.expect("startup bundle digest"),
),
attachment_kind: AttachmentKind::PiSkillPath,
}
}
#[derive(Clone, Copy)]
enum Revocation {
Root,
Custody,
Budget,
}
impl Revocation {
fn error(self) -> &'static str {
match self {
Self::Root => "root session authority moved",
Self::Custody => "review subject is terminally malignant",
Self::Budget => "effective concurrent child",
}
}
fn apply(self, root: &Path, store: &DispatchStore, pending: &PendingDispatch, now: i64) {
match self {
Self::Root => {
let mut next = store
.load_current_root_binding(&pending.root_session_id)
.expect("current root");
next.run = RunId::new("v646").expect("next run");
next.mode = "planning".into();
next.bound_at += 1;
let state: RunState = serde_json::from_value(serde_json::json!({
"run": "v646",
"status": "planted"
}))
.expect("next run state");
state
.store(&store.runs_root().join("v646/run.json"))
.expect("next run");
store
.publish_root_binding_for_run(&next)
.expect("next root binding");
store
.activate_current_root_binding(&next)
.expect("root continuation");
}
Self::Custody => {
let ruling = |revision: u8| ReviewRuling {
schema: "shepherd.review-ruling/1".into(),
project_id: pending.project_id.clone(),
run: pending.run.clone(),
subject_agent_id: pending.expected_attachment.agent_id.clone(),
task_sha256: pending.task_sha256,
task_generation: 1,
reviewer_dispatch_id: DispatchId::new(format!("reviewer-{revision}"))
.expect("reviewer"),
findings_sha256: Sha256::digest(format!("findings-{revision}")).into(),
verdict: ReviewVerdict::Redo,
ruled_at: now + i64::from(revision),
};
let mut custody = ReviewCustody::begin(
&ruling(1),
pending.root_session_id.clone(),
pending.expected_child_session_id.clone(),
pending.role,
pending.lane.clone(),
pending.launch_id_hash,
)
.expect("review custody");
for revision in 1..=4 {
custody
.apply(&ruling(revision))
.expect("adversarial ruling");
}
assert_eq!(custody.state, ReviewCustodyState::Malignant);
store
.publish_review_custody(&custody)
.expect("terminal custody");
}
Self::Budget => {
fs::write(
root.join(".shepherd/shepherd.toml"),
"[spawn]\nmax_parallel = 1\n",
)
.expect("operator reduced the current project capacity");
}
}
}
}
#[derive(Clone, Copy)]
enum LaunchOutcome {
Active,
ClaimRejected(Revocation),
ActivationRejected(Revocation),
}
impl LaunchOutcome {
fn label(self) -> &'static str {
match self {
Self::Active => "activate",
Self::ClaimRejected(Revocation::Root) => "claim-root",
Self::ClaimRejected(Revocation::Custody) => "claim-custody",
Self::ActivationRejected(Revocation::Root) => "activate-root",
Self::ActivationRejected(Revocation::Custody) => "activate-custody",
Self::ClaimRejected(Revocation::Budget) => "claim-budget",
Self::ActivationRejected(Revocation::Budget) => "activate-budget",
}
}
}
#[test]
fn broker_child_process_helper() {
let Some(endpoint) = std::env::var_os("SHEPHERD_BROKER_CHILD_ENDPOINT") else {
return;
};
let error_marker =
PathBuf::from(std::env::var_os("SHEPHERD_BROKER_CHILD_ERROR").expect("error marker"));
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let launch_id = shepherd_cli::BrokerLaunchId::from_opaque(
&std::env::var("SHEPHERD_BROKER_CHILD_LAUNCH").expect("launch id"),
)
.expect("opaque launch id");
let ready =
PathBuf::from(std::env::var_os("SHEPHERD_BROKER_CHILD_READY").expect("ready"));
let claim_go = PathBuf::from(
std::env::var_os("SHEPHERD_BROKER_CHILD_CLAIM_GO").expect("claim go"),
);
let claimed =
PathBuf::from(std::env::var_os("SHEPHERD_BROKER_CHILD_CLAIMED").expect("claimed"));
let activate_go = PathBuf::from(
std::env::var_os("SHEPHERD_BROKER_CHILD_ACTIVATE_GO").expect("activate go"),
);
fs::write(&ready, b"ready").expect("ready marker");
wait_for_file(&claim_go);
let mut client =
shepherd_cli::BrokerClient::connect_child_by_id(endpoint.clone(), launch_id)
.expect("child broker connection");
if std::env::var_os("SHEPHERD_BROKER_CHILD_DUPLICATE").is_some() {
assert!(
shepherd_cli::BrokerClient::connect_child_by_id(endpoint, launch_id).is_err(),
"a second child connection must be rejected"
);
}
let root = PathBuf::from(std::env::var_os("SHEPHERD_BROKER_CHILD_ROOT").expect("root"));
let attestation = loaded_attestation(
&root,
digest_from_hex(&std::env::var("SHEPHERD_BROKER_CHILD_NONCE").expect("nonce")),
);
let claim_result = client.claim(
"conductor-agent".into(),
"conductor-session".into(),
"pi-subagents:conductor".into(),
attestation.clone(),
);
if let Ok(expected) = std::env::var("SHEPHERD_BROKER_CHILD_REJECT_CLAIM") {
let error = claim_result.expect_err("revoked child claim must fail");
assert!(error.to_string().contains(&expected), "{error}");
fs::write(&claimed, b"rejected").expect("claim rejection marker");
wait_for_file(&activate_go);
println!("broker-child-rejected-claim");
return;
}
claim_result.expect("claim child launch");
fs::write(&claimed, b"claimed").expect("claimed marker");
wait_for_file(&activate_go);
let activation = client.activate(
"conductor-agent".into(),
"conductor-session".into(),
"pi-subagents:conductor".into(),
attestation,
);
if let Ok(expected) = std::env::var("SHEPHERD_BROKER_CHILD_REJECT_ACTIVATION") {
let error = activation.expect_err("revoked child activation must fail");
assert!(error.to_string().contains(&expected), "{error}");
println!("broker-child-rejected-activation");
} else {
activation.expect("activate child launch");
println!("broker-child-activated");
}
}));
if let Err(payload) = result {
let detail = payload
.downcast_ref::<&str>()
.copied()
.or_else(|| payload.downcast_ref::<String>().map(String::as_str))
.unwrap_or("child callback failed");
fs::write(error_marker, detail).expect("error marker");
std::panic::resume_unwind(payload);
}
}
#[test]
fn prepare_is_broker_only_and_persists_claimable_state_without_a_secret() {
exercise_broker_launch(LaunchOutcome::Active);
}
#[test]
fn broker_preparation_reserves_project_capacity_before_provider_creation() {
exercise_broker_launch_with_budget(LaunchOutcome::Active, true);
}
#[test]
fn broker_claim_rejects_a_root_that_continued_to_another_run() {
exercise_broker_launch(LaunchOutcome::ClaimRejected(Revocation::Root));
}
#[test]
fn broker_activation_rechecks_root_authority_after_claim() {
exercise_broker_launch(LaunchOutcome::ActivationRejected(Revocation::Root));
}
#[test]
fn broker_claim_rejects_terminal_review_custody() {
exercise_broker_launch(LaunchOutcome::ClaimRejected(Revocation::Custody));
}
#[test]
fn broker_activation_rechecks_review_custody_after_claim() {
exercise_broker_launch(LaunchOutcome::ActivationRejected(Revocation::Custody));
}
#[test]
fn broker_claim_rechecks_reduced_project_capacity() {
exercise_broker_launch(LaunchOutcome::ClaimRejected(Revocation::Budget));
}
#[test]
fn broker_activation_rechecks_reduced_project_capacity() {
exercise_broker_launch(LaunchOutcome::ActivationRejected(Revocation::Budget));
}
fn other_lane(mut request: PreparePendingDispatchRequest) -> PreparePendingDispatchRequest {
request.lane = Some("l2-conductor".into());
request.child_session_id = "excess-session".into();
request.expected_attachment.agent_id = "excess-agent".into();
request.task_file = ".shepherd/runs/v645/lanes/l2-conductor/plan.md".into();
request.read_scope = vec![request.task_file.clone()];
request.result_artifact =
".shepherd/runs/v645/lanes/l2-conductor/reports/conductor.md".into();
request.review_artifact =
".shepherd/runs/v645/lanes/l2-conductor/reviews/conductor.md".into();
request.write_scope = vec![request.result_artifact.clone()];
request
}
fn exercise_broker_launch(outcome: LaunchOutcome) {
exercise_broker_launch_with_budget(outcome, false);
}
fn exercise_broker_launch_with_budget(outcome: LaunchOutcome, check_budget: bool) {
let root = fixture(outcome.label());
fs::create_dir_all(root.join(".shepherd/runs/v645")).expect("runs");
fs::create_dir_all(root.join("docs")).expect("docs");
if check_budget {
fs::write(
root.join(".shepherd/shepherd.toml"),
"[spawn]\nmax_parallel = 1\n",
)
.expect("one native process slot");
}
fs::write(
root.join(".shepherd/project.json"),
br#"{"id":"018f47ce-72d7-7f64-9eb1-2f651d521c2a"}"#,
)
.expect("project");
fs::write(root.join("docs/task.md"), b"task\n").expect("task");
let installed = root.join("installed");
let compiled = Command::new(env!("CARGO_BIN_EXE_shepherd"))
.args(["compile", "--target", "pi", "--out"])
.arg(&installed)
.current_dir(&root)
.output()
.expect("compile installed carrier");
assert!(
compiled.status.success(),
"compiler stderr={}",
String::from_utf8_lossy(&compiled.stderr)
);
fs::write(root.join(".gitignore"), b"installed/\n").expect("gitignore");
assert!(
Command::new("git")
.args(["init", "--quiet"])
.current_dir(&root)
.status()
.expect("git init")
.success()
);
let baseline = super::plan_support::open_execution(
&root,
"v645",
&["l1-conductor", "l2-conductor"],
if check_budget { 1 } else { 4 },
);
let now = i64::try_from(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("clock")
.as_millis(),
)
.expect("millisecond clock fits i64");
let registry_path = root.join(".shepherd/shepherd.db");
let registry = Registry::open_migrated(®istry_path).expect("registry");
registry
.execute(
"INSERT INTO projects (id, name, created_at, updated_at) VALUES (?1, ?2, ?3, ?3) ON CONFLICT(id) DO NOTHING",
(
"018f47ce-72d7-7f64-9eb1-2f651d521c2a",
"broker fixture",
now,
),
)
.expect("seed registry project");
let store = DispatchStore::new(root.join(".shepherd/runs"));
let service = DispatchService::with_project_root(
store.clone(),
ProjectId::new("018f47ce-72d7-7f64-9eb1-2f651d521c2a").expect("project"),
&root,
)
.with_installed_package(&installed, installed.join(".shepherd-generated.json"));
service
.bind_root(
shepherd_cli::BindRootDispatchRequest {
schema: "shepherd.dispatch-request/1".into(),
run: Some("v645".into()),
harness: Harness::Pi,
session_id: "root-session".into(),
role_carrier: "shepherd:shepherd".into(),
mode: "execution".into(),
lease_ms: 600_000,
},
now,
)
.unwrap_or_else(|error| panic!("root bind root={} error={error}", root.display()));
let endpoint = broker_endpoint(outcome.label());
let broker = NativeBroker::start(service, &endpoint).expect("broker");
let mut parent = broker.connect().expect("parent connection");
parent
.register_parent(
Harness::Pi,
shepherd_cli::shepherd::dispatch::Role::Shepherd,
shepherd_cli::shepherd::dispatch::SessionId::new("root-session").expect("session"),
shepherd_cli::shepherd::dispatch::SessionId::new("root-session").expect("root"),
None,
)
.expect("parent registration");
let prepare_request = PreparePendingDispatchRequest {
schema: "shepherd.pending-dispatch-request/2".into(),
run: Some("v645".into()),
role: "conductor".into(),
work_kind: "coordination".into(),
lane: Some("l1-conductor".into()),
parent_dispatch_id: None,
replaces_agent_id: None,
baseline,
read_scope: vec![".shepherd/runs/v645/lanes/l1-conductor/plan.md".into()],
write_scope: vec![".shepherd/runs/v645/lanes/l1-conductor/reports/conductor.md".into()],
result_artifact: ".shepherd/runs/v645/lanes/l1-conductor/reports/conductor.md".into(),
review_artifact: ".shepherd/runs/v645/lanes/l1-conductor/reviews/conductor.md".into(),
task_file: ".shepherd/runs/v645/lanes/l1-conductor/plan.md".into(),
child_session_id: "conductor-session".into(),
lease_ms: 60_000,
expected_attachment: CarrierAttachmentExpectationRequest {
target: Harness::Pi,
role: "conductor".into(),
agent_id: "conductor-agent".into(),
attachment_kind: "pi-skill-path".into(),
},
};
let mut wrong_task = prepare_request.clone();
wrong_task.task_file = "docs/task.md".into();
wrong_task.read_scope = vec!["docs/**".into()];
let error = parent.prepare(wrong_task).expect_err(
"a valid Conductor identity cannot substitute an arbitrary brief for the lane plan",
);
assert!(
error.to_string().contains("exact rendered lane plan"),
"{error}"
);
let launch = parent
.prepare(prepare_request.clone())
.expect("prepare through broker");
if check_budget {
let excess = other_lane(prepare_request.clone());
let error = parent.prepare(excess).expect_err(
"a pending reservation must consume capacity before any provider exists",
);
assert!(
error.to_string().contains("effective concurrent child"),
"{error}"
);
assert!(
!root
.join(".shepherd/runs/v645/dispatch/excess-agent.json")
.exists()
);
}
assert!(format!("{launch:?}").contains("launch_id: \"<opaque>\""));
let prepared = store
.load_pending_for_agent(
&RunId::new("v645").expect("run"),
&AgentId::new("conductor-agent").expect("agent"),
)
.expect("prepared Conductor reservation");
assert_eq!(prepared.nonce_sha256, launch.nonce_sha256());
let pending = root.join(format!(
".shepherd/runs/v645/dispatch/pending-{}.json",
hex(&prepared.launch_id_hash)
));
let value: serde_json::Value =
serde_json::from_slice(&fs::read(&pending).expect("pending bytes"))
.expect("pending json");
assert_eq!(value["launch_state"], "pending");
assert!(value.get("secret").is_none());
assert_eq!(value["nonce_sha256"], hex(&launch.nonce_sha256()));
let pending_record: PendingDispatch =
serde_json::from_value(value).expect("pending record");
if matches!(
outcome,
LaunchOutcome::ClaimRejected(Revocation::Budget)
| LaunchOutcome::ActivationRejected(Revocation::Budget)
) {
parent
.prepare(other_lane(prepare_request))
.expect("second reservation fits the initial project capacity");
}
let ready = root.join("child-ready");
let claim_go = root.join("child-claim-go");
let claimed = root.join("child-claimed");
let activate_go = root.join("child-activate-go");
let child_error = root.join("child-error");
let child_executable = std::env::current_exe().expect("test executable");
let mut child_command = Command::new(child_executable);
child_command
.args([
"--exact",
"unix::broker_child_process_helper",
"--nocapture",
])
.env("SHEPHERD_BROKER_CHILD_ENDPOINT", &endpoint)
.env(
"SHEPHERD_BROKER_CHILD_LAUNCH",
launch.launch_id().opaque_string(),
)
.env("SHEPHERD_BROKER_CHILD_ROOT", &installed)
.env("SHEPHERD_BROKER_CHILD_READY", &ready)
.env("SHEPHERD_BROKER_CHILD_CLAIM_GO", &claim_go)
.env("SHEPHERD_BROKER_CHILD_CLAIMED", &claimed)
.env("SHEPHERD_BROKER_CHILD_ACTIVATE_GO", &activate_go)
.env("SHEPHERD_BROKER_CHILD_ERROR", &child_error)
.env("SHEPHERD_BROKER_CHILD_DUPLICATE", "1")
.env("SHEPHERD_BROKER_CHILD_NONCE", hex(&launch.nonce_sha256()))
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
match outcome {
LaunchOutcome::Active => {}
LaunchOutcome::ClaimRejected(revocation) => {
child_command.env("SHEPHERD_BROKER_CHILD_REJECT_CLAIM", revocation.error());
}
LaunchOutcome::ActivationRejected(revocation) => {
child_command.env(
"SHEPHERD_BROKER_CHILD_REJECT_ACTIVATION",
revocation.error(),
);
}
}
let child = child_command.spawn().expect("spawn child");
wait_for_file(&ready);
assert!(parent.register_child(&launch, std::process::id()).is_err());
parent
.register_child(&launch, child.id())
.expect("register child");
assert!(
shepherd_cli::BrokerClient::connect_child_by_id(&endpoint, launch.launch_id()).is_err(),
"the registered parent PID cannot impersonate the child connection"
);
if let LaunchOutcome::ClaimRejected(revocation) = outcome {
revocation.apply(&root, &store, &pending_record, now);
}
fs::write(&claim_go, b"go").expect("claim go");
wait_for_claim(&claimed, &child_error);
let claimed_value: serde_json::Value =
serde_json::from_slice(&fs::read(&pending).expect("claimed pending bytes"))
.expect("claimed pending json");
assert_eq!(
claimed_value["launch_state"],
if matches!(outcome, LaunchOutcome::ClaimRejected(_)) {
"pending"
} else {
"claimed-unspawned"
}
);
assert!(
!root
.join(".shepherd/runs/v645/dispatch/conductor-agent.json")
.exists()
);
if let LaunchOutcome::ActivationRejected(revocation) = outcome {
revocation.apply(&root, &store, &pending_record, now);
}
fs::write(&activate_go, b"go").expect("activate or disconnect go");
let output = child.wait_with_output().expect("child output");
let expected_marker = match outcome {
LaunchOutcome::Active => "broker-child-activated",
LaunchOutcome::ClaimRejected(_) => "broker-child-rejected-claim",
LaunchOutcome::ActivationRejected(_) => "broker-child-rejected-activation",
};
assert!(
output.status.success()
&& String::from_utf8_lossy(&output.stdout).contains(expected_marker),
"stdout={} stderr={}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
let expected_state = match outcome {
LaunchOutcome::Active => "active",
LaunchOutcome::ClaimRejected(_) | LaunchOutcome::ActivationRejected(_) => {
"launch-failed"
}
};
let active_value = wait_for_pending_state(&pending, expected_state);
assert_eq!(active_value["launch_state"], expected_state);
assert_eq!(
root.join(".shepherd/runs/v645/dispatch/conductor-agent.json")
.is_file(),
matches!(outcome, LaunchOutcome::Active),
);
let registry = Registry::open_migrated(®istry_path).expect("reopen registry");
let claim = registry
.load_dispatch_singleton(
"018f47ce-72d7-7f64-9eb1-2f651d521c2a",
"v645",
"conductor",
"l1-conductor",
)
.expect("load singleton");
if matches!(outcome, LaunchOutcome::Active) {
let claim = claim.expect("production activation must publish singleton claim");
assert_eq!(claim.agent_id, "conductor-agent");
assert_eq!(
claim.publication_state,
Some(SingletonPublicationState::Published)
);
assert!(claim.publication_nonce.is_some());
} else {
assert!(
claim.is_none(),
"revoked child must not acquire the singleton"
);
}
assert!(
shepherd_cli::BrokerClient::connect_child_by_id(&endpoint, launch.launch_id()).is_err()
);
drop(parent);
drop(broker);
fs::remove_dir_all(root).expect("cleanup");
fs::remove_dir_all(endpoint.parent().expect("broker parent")).expect("broker cleanup");
}
#[test]
fn private_broker_authenticates_parent_without_reading_a_descriptor_file() {
let root = fixture("private");
let service = shepherd_cli::DispatchService::new(
DispatchStore::new(root.join("runs")),
ProjectId::new("018f47ce-72d7-7f64-9eb1-2f651d521c2a").expect("project"),
);
let endpoint = broker_endpoint("private");
let broker = NativeBroker::start(service, &endpoint).expect("broker");
let mut client = broker.connect().expect("connect");
client
.register_parent(
Harness::Pi,
shepherd_cli::shepherd::dispatch::Role::Shepherd,
shepherd_cli::shepherd::dispatch::SessionId::new("root-session").expect("session"),
shepherd_cli::shepherd::dispatch::SessionId::new("root-session").expect("root"),
None,
)
.expect("registered parent");
assert!(
fs::symlink_metadata(&endpoint)
.expect("socket metadata")
.file_type()
.is_socket()
);
assert!(ProcessIdentity::current().expect("identity").pid() > 0);
drop(client);
drop(broker);
assert!(!endpoint.exists());
fs::remove_dir_all(endpoint.parent().expect("broker parent")).expect("broker cleanup");
fs::remove_dir_all(root).expect("cleanup");
}
#[test]
fn broker_refuses_a_path_that_is_already_a_json_file() {
let root = fixture("forged-json");
let endpoint = root.join("broker.sock");
fs::write(&endpoint, br#"{"fake":"authority"}"#).expect("forged file");
let service = shepherd_cli::DispatchService::new(
DispatchStore::new(root.join("runs")),
ProjectId::new("018f47ce-72d7-7f64-9eb1-2f651d521c2a").expect("project"),
);
let error = match NativeBroker::start(service, &endpoint) {
Ok(_) => panic!("json is not a socket"),
Err(error) => error,
};
assert!(matches!(error, BrokerError::Protocol(_)));
fs::remove_dir_all(root).expect("cleanup");
}
}