use std::{
collections::BTreeSet,
fs,
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, PathAuthority,
PendingDispatch, PendingLaunchState, ProjectFilesystemId, ProjectId, ROOT_SESSION_SCHEMA,
ReviewCustody, ReviewCustodyState, ReviewRuling, ReviewVerdict, Role, RootSessionBinding,
RunId, SessionId,
},
};
use shepherd_cli::{DispatchStore, DispatchStoreError, ReviewQuarantineFault, RunStore};
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-dispatch-store-{label}-{suffix:x}"));
fs::create_dir_all(&path).expect("fixture");
fs::canonicalize(path).expect("canonical fixture")
}
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(),
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)
}
#[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]
);
fs::remove_dir_all(root).expect("cleanup");
}
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],
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"),
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"
);
fs::remove_dir_all(root).expect("cleanup");
}
}
#[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());
fs::remove_dir_all(root).expect("cleanup");
}
#[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);
fs::remove_dir_all(root).expect("cleanup");
}
#[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");
fs::remove_dir_all(root).expect("cleanup");
}
#[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 { .. }));
fs::remove_dir_all(root).expect("cleanup");
}
#[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()
);
fs::remove_dir_all(root).expect("cleanup");
}
#[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);
fs::remove_dir_all(root).expect("cleanup");
}
#[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());
fs::remove_dir_all(root).expect("cleanup");
}
#[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());
fs::remove_dir_all(root).expect("cleanup");
}
#[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 { .. }));
fs::remove_dir_all(root).expect("cleanup");
}
#[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");
fs::remove_dir_all(root).expect("cleanup");
}
#[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");
fs::remove_dir_all(root).expect("cleanup");
}