#![cfg(unix)]
#[path = "support/broker.rs"]
mod broker_fixture;
#[path = "support/plan.rs"]
mod plan_fixture;
use std::{
fs,
path::Path,
process::Command,
thread,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use sha2::{Digest, Sha256};
use shepherd_cli::shepherd::dispatch::{
GitCommit, MutationCheckpoint, MutationCheckpointFile, MutationCheckpointIdentity,
MutationCheckpointState, ProjectId, Role, RootMode,
};
use shepherd_cli::{
BindRootDispatchRequest, BrokerClient, CarrierAttachmentExpectationRequest, DispatchService,
DispatchStore, Harness, NativeBroker, PreparePendingDispatchRequest, RunStore,
};
fn sha256(bytes: &[u8]) -> [u8; 32] {
Sha256::digest(bytes).into()
}
fn wait_for(path: &Path) {
let deadline = std::time::Instant::now() + Duration::from_secs(30);
while !path.exists() {
assert!(
std::time::Instant::now() < deadline,
"timed out waiting for {}",
path.display()
);
thread::sleep(Duration::from_millis(5));
}
}
#[test]
fn prepared_high_blast_launch_rejects_live_target_drift() {
prepared_high_blast_launch_rejects(MutationCase::LiveTarget);
}
#[test]
fn prepared_high_blast_launch_rejects_snapshot_drift() {
prepared_high_blast_launch_rejects(MutationCase::Snapshot);
}
#[test]
fn prepared_high_blast_launch_rejects_receipt_drift() {
prepared_high_blast_launch_rejects(MutationCase::Receipt);
}
#[derive(Clone, Copy)]
enum MutationCase {
LiveTarget,
Snapshot,
Receipt,
}
fn prepared_high_blast_launch_rejects(case: MutationCase) {
let guard = tempfile::Builder::new()
.prefix("shepherd-mutation-e2e-")
.tempdir()
.expect("fixture root");
let root = fs::canonicalize(guard.path()).expect("canonical root");
fs::create_dir_all(root.join("docs")).expect("docs");
fs::create_dir_all(root.join(".shepherd")).expect("shepherd directory");
plan_fixture::write_project_identity(&root, "018f47ce-72d7-7f64-9eb1-2f651d521c2a", 1000);
let external = tempfile::tempdir().expect("external custody");
let snapshot_root = external.path().join("snapshot");
let receipt_path = external.path().join("receipt.json");
let task = format!(
"mutation_risk: high-blast\nrecovery_checkpoint: {}\n",
receipt_path.display()
);
fs::write(root.join("docs/task.md"), task.as_bytes()).expect("task");
let live_path = root.join("docs/live.txt");
let original = b"before mutation\n";
fs::write(&live_path, original).expect("live target");
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 carrier");
assert!(
compiled.status.success(),
"compiler failed: {}",
String::from_utf8_lossy(&compiled.stderr)
);
let git = |args: &[&str]| {
let output = Command::new("git")
.args(args)
.current_dir(&root)
.output()
.expect("git command");
assert!(
output.status.success(),
"git {:?}: {}",
args,
String::from_utf8_lossy(&output.stderr)
);
};
git(["init", "--quiet"].as_slice());
git(["config", "user.email", "fixture@example.invalid"].as_slice());
git(["config", "user.name", "Fixture"].as_slice());
let baseline = plan_fixture::open_execution(&root, "v670", &["lane-a"], 2);
let project = ProjectId::new("018f47ce-72d7-7f64-9eb1-2f651d521c2a").expect("project id");
let store = DispatchStore::new(root.join(".shepherd/runs"));
let service = DispatchService::with_project_root(store, project.clone(), root.clone())
.with_installed_package(&installed, installed.join(".shepherd-generated.json"));
let now = i64::try_from(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("clock")
.as_millis(),
)
.expect("clock fits i64");
service
.bind_root(
BindRootDispatchRequest {
schema: "shepherd.dispatch-request/1".into(),
run: Some("v670".into()),
harness: Harness::Pi,
session_id: "root-session".into(),
role_carrier: "shepherd:shepherd".into(),
mode: RootMode::Execution,
lease_ms: 600_000,
},
now,
)
.expect("bind root");
let state = RunStore::new(root.join(".shepherd/runs/v670/run.json"))
.load()
.expect("run state");
let project_filesystem_id = service.project_filesystem_id().expect("filesystem id");
fs::create_dir_all(snapshot_root.join("docs")).expect("snapshot parent");
fs::write(snapshot_root.join("docs/live.txt"), original).expect("snapshot");
let metadata = fs::symlink_metadata(&live_path).expect("live metadata");
use std::os::unix::fs::MetadataExt;
let identity = MutationCheckpointIdentity {
device: metadata.dev(),
inode: metadata.ino(),
nlink: metadata.nlink(),
};
let created_at = now / 1_000;
let checkpoint = MutationCheckpoint {
schema: shepherd_cli::shepherd::dispatch::MUTATION_CHECKPOINT_SCHEMA.into(),
project_id: project.clone(),
project_filesystem_id,
run: shepherd_cli::shepherd::dispatch::RunId::new(state.run.clone()).expect("run id"),
run_incarnation: state.run_incarnation.clone(),
branch: state.branch.clone(),
base: state.base.clone(),
base_commit: GitCommit::new(baseline.clone()).expect("base commit"),
lane: "lane-a".into(),
baseline: GitCommit::new(baseline.clone()).expect("baseline"),
task_sha256: sha256(task.as_bytes()),
worktree_root: root.to_string_lossy().replace('\\', "/"),
snapshot_root: fs::canonicalize(&snapshot_root)
.expect("snapshot path")
.to_string_lossy()
.replace('\\', "/"),
created_at,
files: vec![MutationCheckpointFile {
path: "docs/live.txt".into(),
state: MutationCheckpointState::Regular,
sha256: Some(sha256(original)),
bytes: original.len() as u64,
mode: metadata.mode() & 0o7777,
identity: Some(identity),
}],
};
fs::write(
&receipt_path,
serde_json::to_vec(&checkpoint).expect("receipt JSON"),
)
.expect("receipt");
let endpoint_dir = tempfile::tempdir_in("/tmp").expect("short broker endpoint dir");
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
fs::set_permissions(endpoint_dir.path(), fs::Permissions::from_mode(0o700))
.expect("private endpoint directory");
}
let endpoint = endpoint_dir.path().join("broker.sock");
let broker = NativeBroker::start(service, &endpoint).expect("native broker");
let mut parent = broker.connect().expect("parent connection");
parent
.register_parent(
Harness::Pi,
Role::Shepherd,
shepherd_cli::shepherd::dispatch::SessionId::new("root-session").expect("session"),
shepherd_cli::shepherd::dispatch::SessionId::new("root-session").expect("root"),
None,
)
.expect("register parent");
let request = PreparePendingDispatchRequest {
schema: "shepherd.pending-dispatch-request/2".into(),
run: Some("v670".into()),
role: "worker".into(),
work_kind: "artifact".into(),
lane: Some("lane-a".into()),
parent_dispatch_id: None,
replaces_agent_id: None,
baseline,
read_scope: vec!["docs/task.md".into()],
write_scope: vec!["docs/live.txt".into()],
result_artifact: ".shepherd/runs/v670/lanes/lane-a/workers/worker.json".into(),
review_artifact: ".shepherd/runs/v670/lanes/lane-a/reviews/worker.json".into(),
task_file: "docs/task.md".into(),
child_session_id: "worker-session".into(),
lease_ms: 60_000,
expected_attachment: CarrierAttachmentExpectationRequest {
target: Harness::Pi,
role: "worker".into(),
agent_id: "worker-agent".into(),
attachment_kind: "pi-skill-path".into(),
},
};
let launch = parent
.prepare(request.clone())
.expect("prepare high-blast launch");
let child_endpoint = endpoint.clone();
let child_installed = installed.clone();
let child_scratch = external.path().join("provider");
let child_scratch_thread = child_scratch.clone();
let child_request = request.clone();
let child = thread::spawn(move || {
let mut child_parent =
BrokerClient::connect_endpoint(&child_endpoint).expect("child parent");
child_parent
.register_parent(
Harness::Pi,
Role::Shepherd,
shepherd_cli::shepherd::dispatch::SessionId::new("root-session").expect("session"),
shepherd_cli::shepherd::dispatch::SessionId::new("root-session").expect("root"),
None,
)
.expect("register child parent");
let result = broker_fixture::LiveProvider::launch_prepared(
&mut child_parent,
&child_endpoint,
&child_installed,
&child_scratch_thread,
child_request,
&launch,
);
assert!(result.is_err(), "mutated checkpoint must reject activation");
});
wait_for(&child_scratch.join("worker-agent/ready"));
match case {
MutationCase::LiveTarget => {
fs::write(&live_path, b"drifted live target\n").expect("mutate live target");
}
MutationCase::Snapshot => {
fs::write(snapshot_root.join("docs/live.txt"), b"drifted snapshot\n")
.expect("mutate snapshot");
}
MutationCase::Receipt => {
fs::write(&receipt_path, b"{}\n").expect("mutate receipt");
}
}
child.join().expect("child process");
drop(parent);
drop(broker);
}
#[test]
fn broker_fixture_child() {
broker_fixture::child_main();
}