//! Real native broker lifecycle with deterministic child report content.
//! No dispatch, pending, singleton, or orientation authority is hand-authored.
#![cfg(unix)]
#[path = "support/broker.rs"]
mod broker_fixture;
#[path = "support/plan.rs"]
#[allow(dead_code)]
mod plan_fixture;
use shepherd_cli::shepherd::run::{LaneStatus, RunStatus, Vocabulary};
use std::{
fs,
os::unix::fs::PermissionsExt,
path::Path,
process::Command,
time::{SystemTime, UNIX_EPOCH},
};
use sha2::{Digest, Sha256};
use shepherd_cli::{
BindRootDispatchRequest, CarrierAttachmentExpectationRequest, DispatchService, DispatchStore,
NativeBroker, PreparePendingDispatchRequest, ReviewReplacementRequest, ReviewRulingRequest,
shepherd::{
Harness, RunState,
dispatch::{
DispatchRecord, DispatchState, ProjectId, ReviewCustodyState, Role, RunId, SessionId,
},
registry::Registry,
},
};
const RUN: &str = "v913";
const ROOT_SESSION: &str = "orientation-root";
const ENGINEER: &str = "planning-engineer";
fn hash(bytes: &[u8]) -> String {
Sha256::digest(bytes)
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
}
fn now() -> i64 {
i64::try_from(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("clock")
.as_millis(),
)
.expect("bounded clock")
}
fn accepted(root: &Path, args: &[&str]) -> std::process::Output {
let output = plan_fixture::invoke(root, args);
assert!(
output.status.success(),
"{args:?}: {}",
String::from_utf8_lossy(&output.stderr)
);
output
}
fn request(root: &Path, role: Role) -> PreparePendingDispatchRequest {
let id = format!("planning-{role}");
let prefix = format!(".shepherd/runs/{RUN}");
let result = format!(
"{prefix}/reports/{id}.{}",
if role == Role::Engineer { "md" } else { "json" }
);
let baseline = Command::new("git")
.args(["rev-parse", "HEAD"])
.current_dir(root)
.output()
.expect("baseline");
assert!(baseline.status.success());
PreparePendingDispatchRequest {
schema: "shepherd.pending-dispatch-request/2".into(),
run: Some(RUN.into()),
role: format!("shepherd:{role}"),
work_kind: match role {
Role::Engineer => "planning",
Role::Discovery => "research",
_ => "review",
}
.into(),
lane: None,
parent_dispatch_id: (role != Role::Engineer).then(|| ENGINEER.into()),
replaces_agent_id: None,
baseline: String::from_utf8(baseline.stdout)
.expect("baseline UTF-8")
.trim()
.into(),
read_scope: {
let mut scope = vec![
"docs/task.md".into(),
"src/fact.txt".into(),
format!("{prefix}/seed.md"),
];
if role == Role::Critic {
scope.extend(
[
"mesh.md",
"phase0.md",
"plan.md",
"plan-probes.json",
"graph/topology.json",
"lanes/lane-a/plan.md",
]
.map(|relative| format!("{prefix}/{relative}")),
);
}
scope.sort();
scope
},
write_scope: if role == Role::Engineer {
vec![
format!("{prefix}/phase0.md"),
format!("{prefix}/plan.md"),
result.clone(),
]
} else {
vec![]
},
result_artifact: result,
review_artifact: format!("{prefix}/reviews/{id}.json"),
task_file: "docs/task.md".into(),
child_session_id: format!("session-{id}"),
lease_ms: 300_000,
expected_attachment: CarrierAttachmentExpectationRequest {
target: Harness::Pi,
role: format!("shepherd:{role}"),
agent_id: id,
attachment_kind: "pi-skill-path".into(),
},
}
}
fn lineage_request(root: &Path, role: Role, prefix: &str) -> PreparePendingDispatchRequest {
let mut value = request(root, role);
if prefix != "planning" {
let id = format!("{prefix}-{role}");
value.expected_attachment.agent_id = id.clone();
value.child_session_id = format!("session-{id}");
if role == Role::Engineer {
value.replaces_agent_id = Some(ENGINEER.into());
} else {
value.parent_dispatch_id = Some(format!("{prefix}-engineer"));
value.result_artifact = format!(".shepherd/runs/{RUN}/reports/{id}.json");
value.review_artifact = format!(".shepherd/runs/{RUN}/reviews/{id}.json");
}
}
value
}
fn write_report(root: &Path, record: &DispatchRecord) {
let id = record.agent_id.to_string();
let evidence = serde_json::json!({"path":"src/fact.txt", "sha256":hash(&fs::read(root.join("src/fact.txt")).expect("evidence")), "line":1});
let mut report = serde_json::json!({"schema":"shepherd.orientation-report/1", "run":RUN,
"result_id":id, "kind":record.role.as_str(), "role":record.role.as_str(), "read_scope":["src/fact.txt"],
"status":"complete", "summary":"Deterministic native lifecycle fixture report.",
"assumptions":[{"id":format!("{id}-assumption"),"statement":"Fixture data only."}],
"claims":[{"id":format!("{id}-claim"),"statement":"Observed the exact fixture file.","evidence":[evidence]}],
"evidence":[evidence], "caveats":[]});
if record.role == Role::Critic {
if root.join(format!(".shepherd/runs/{RUN}/plan.md")).is_file() {
plan_fixture::add_planning_evidence(root, RUN, &mut report);
}
report["orientation_pre_sha256"] = hash(
&fs::read(root.join(format!(".shepherd/runs/{RUN}/orientation-pre.json")))
.expect("native pre"),
)
.into();
let prefix = if id.starts_with("replacement-") {
"replacement"
} else {
"planning"
};
report["verdict"] = serde_json::json!({"decision":"GREEN", "findings":[], "corrections":[], "blockers":[],
"citations":[{"claim_id":format!("{prefix}-auditor-claim")}]});
}
fs::write(
root.join(record.result_artifact.as_ref().expect("native result path")),
serde_json::to_vec(&report).expect("report JSON"),
)
.expect("child report");
}
fn write_phase0(root: &Path, prefix: &str) {
let seed_path = format!(".shepherd/runs/{RUN}/seed.md");
let phase0 = format!(
"# Orientation\n\n## Run and seed\n- run: {RUN}\n- verified seed path: {seed_path}\n- seed hash: {}\n- planted observation: status=planted\n\n## Auditor briefs\n### planning-auditor\n- exact read scope: src/fact.txt\n- output path: reports/planning-auditor.json\n\n## Discovery briefs\n### planning-discovery\n- exact read scope: src/fact.txt\n- output path: reports/planning-discovery.json\n\n## Assumptions and decisions\n- Deterministic content, real native activation.\n\n## Coverage map\n- native-validator: exact plan boundary.\n\n## Self-review\n- Native gates decide acceptance.\n\n## Critic loop\n- Native pre precedes Critic launch.\n",
hash(&fs::read(root.join(&seed_path)).expect("seed"))
);
let phase0 = phase0
.replace("planning-auditor", &format!("{prefix}-auditor"))
.replace("planning-discovery", &format!("{prefix}-discovery"));
fs::write(root.join(format!(".shepherd/runs/{RUN}/phase0.md")), phase0)
.expect("Engineer phase0");
}
#[test]
fn broker_fixture_child() {
broker_fixture::child_main();
}
#[test]
fn active_native_engineer_completes_orientation_review_and_opens_typed_plan() {
exercise_native_orientation(false);
}
#[test]
fn authorized_malignant_replacement_gets_new_pre_epoch_with_immutable_prior_inputs() {
exercise_native_orientation(true);
}
fn exercise_native_orientation(replace: bool) {
let path = std::env::temp_dir().join(format!(
"shepherd-planning-broker-{}-{replace}-{}",
std::process::id(),
now()
));
fs::create_dir_all(&path).expect("fixture root");
fs::set_permissions(&path, fs::Permissions::from_mode(0o700)).expect("private fixture");
let root = fs::canonicalize(path).expect("canonical fixture");
for directory in ["src", "tests", "docs"] {
fs::create_dir_all(root.join(directory)).expect("fixture directory");
}
fs::write(root.join("src/fact.txt"), b"Native lifecycle evidence.\n").expect("evidence file");
fs::write(
root.join("docs/task.md"),
b"Verify exact native planning custody.\n",
)
.expect("task");
fs::write(
root.join(".gitignore"),
b"installed/\nprovider/\nisolated-home/\n",
)
.expect("gitignore");
assert!(
Command::new("git")
.args(["init", "--quiet"])
.current_dir(&root)
.status()
.expect("git init")
.success()
);
accepted(&root, &["init", "--confirm"]);
accepted(&root, &["run", "init", RUN]);
plan_fixture::seed(&root, RUN);
for directory in ["reports", "reviews"] {
fs::create_dir_all(root.join(format!(".shepherd/runs/{RUN}/{directory}")))
.expect("issued artifact directory");
}
let installed = root.join("installed");
accepted(
&root,
&[
"compile",
"--target",
"pi",
"--out",
installed.to_str().expect("UTF-8 root"),
],
);
assert!(
Command::new("git")
.args(["add", "."])
.current_dir(&root)
.status()
.expect("git add")
.success()
);
assert!(
Command::new("git")
.args([
"-c",
"user.name=Fixture",
"-c",
"user.email=fixture@example.test",
"commit",
"-qm",
"native planning fixture"
])
.current_dir(&root)
.status()
.expect("commit fixture")
.success()
);
let identity: serde_json::Value =
serde_json::from_slice(&fs::read(root.join(".shepherd/project.json")).expect("project"))
.expect("project JSON");
let project_id = identity["id"].as_str().expect("project ID");
Registry::open_migrated(root.join(".shepherd/shepherd.db")).expect("registry").execute(
"INSERT INTO projects (id,name,created_at,updated_at) VALUES (?1,?2,?3,?3) ON CONFLICT(id) DO NOTHING",
(project_id, "native orientation fixture", now())).expect("registry project");
let service = DispatchService::with_project_root(
DispatchStore::new(root.join(".shepherd/runs")),
ProjectId::new(project_id).expect("project"),
&root,
)
.with_installed_package(&installed, installed.join(".shepherd-generated.json"));
service
.bind_root(
BindRootDispatchRequest {
schema: "shepherd.dispatch-request/1".into(),
run: Some(RUN.into()),
harness: Harness::Pi,
session_id: ROOT_SESSION.into(),
role_carrier: "shepherd:shepherd".into(),
mode: shepherd_cli::shepherd::dispatch::RootMode::Planting,
lease_ms: 600_000,
},
now(),
)
.expect("native planning root");
let endpoint = fs::canonicalize("/tmp")
.expect("short socket root")
.join(format!("sp-{}-{}", std::process::id(), now()))
.join("broker.sock");
let broker = NativeBroker::start(service.clone(), &endpoint).expect("native broker");
let mut parent = broker.connect().expect("root channel");
parent
.register_parent(
Harness::Pi,
Role::Shepherd,
SessionId::new(ROOT_SESSION).expect("root"),
SessionId::new(ROOT_SESSION).expect("root"),
None,
)
.expect("native root peer");
let mut engineer = broker_fixture::LiveProvider::launch(
&mut parent,
&endpoint,
&installed,
&root.join("provider"),
request(&root, Role::Engineer),
)
.expect("real Engineer activation");
write_phase0(&root, "planning");
assert!(
!plan_fixture::invoke(&root, &["run", "orientation", "pre", RUN])
.status
.success(),
"empty child inventory cannot pass pre"
);
assert!(
engineer.spawn(request(&root, Role::Critic)).is_err(),
"Critic cannot launch before native pre"
);
for role in [Role::Auditor, Role::Discovery] {
let child = engineer
.spawn(request(&root, role))
.expect("actual Engineer child activation");
write_report(&root, &child);
engineer
.complete_child(child.agent_id.as_str())
.expect("native child completion");
}
accepted(&root, &["run", "orientation", "pre", RUN, "--json"]);
let mut prefix = "planning";
if replace {
let original_pre =
fs::read(root.join(format!(".shepherd/runs/{RUN}/orientation-pre.json")))
.expect("first pre");
let mut earlier_request = request(&root, Role::Critic);
earlier_request.expected_attachment.agent_id = "earlier-critic".into();
earlier_request.child_session_id = "earlier-critic-session".into();
earlier_request.result_artifact =
format!(".shepherd/runs/{RUN}/reports/earlier-critic.json");
let earlier = engineer
.spawn(earlier_request)
.expect("earlier real Critic");
write_report(&root, &earlier);
engineer
.complete_child(earlier.agent_id.as_str())
.expect("earlier real Critic completion");
accepted(&root, &["run", "orientation", "post", RUN, "--json"]);
let original_post =
fs::read(root.join(format!(".shepherd/runs/{RUN}/orientation-post.json")))
.expect("old post retained");
let old_critic = engineer
.spawn(request(&root, Role::Critic))
.expect("first real Critic");
let original = engineer.record().clone();
let baseline = request(&root, Role::Engineer).baseline;
for round in 1..=4 {
let ruling: ReviewRulingRequest = serde_json::from_value(serde_json::json!({
"schema":"shepherd.review-ruling-request/1", "run":RUN, "harness":"pi",
"root_session_id":ROOT_SESSION, "subject_agent_id":ENGINEER,
"reviewer_dispatch_id":old_critic.agent_id, "reviewer_session_id":old_critic.session_id, "task_generation":1,
"review":{"schema":"shepherd.review-result/1", "run":RUN, "lane":null,
"mode":"critic-prehoc", "reviewer_role":"critic", "candidate_commit":baseline,
"input_digest":"aa".repeat(32), "startup_skill":"reviewing", "skill_bundle_digest":"bb".repeat(32),
"result_channel":"native-result", "verdict":"red", "report_path":null,
"findings":[{"finding_id":format!("orientation-failure-{round}"), "location":"docs/task.md:1",
"hypothesis":"Deterministic planning acceptance is missing", "falsification_command":"cargo test",
"falsification_exit_status":1, "observed_result":format!("fixture failure {round}"),
"confidence":"structurally-verifiable", "severity":"important", "impact":"fixture outcome blocked",
"acceptance_predicate":"native orientation fixture passes", "owner_role":"engineer", "route":"redo subject",
"evidence_paths":[format!("fixture/failure-{round}.txt")]}]}
})).expect("typed fixture ruling");
let custody = service
.review_ruling(ruling, now())
.expect("native review ruling");
assert_eq!(custody.rejected_revisions, round);
assert_eq!(
custody.state,
if round == 4 {
ReviewCustodyState::Malignant
} else {
ReviewCustodyState::Active
}
);
}
assert!(
!plan_fixture::invoke(&root, &["run", "orientation", "pre", RUN])
.status
.success(),
"malignant lead cannot reset pre"
);
let next = lineage_request(&root, Role::Engineer, "replacement");
let handle = parent
.prepare(next.clone())
.expect("prepare exact replacement");
service
.review_replace(
ReviewReplacementRequest {
schema: "shepherd.review-replacement-request/1".into(),
run: RUN.into(),
harness: Harness::Pi,
root_session_id: ROOT_SESSION.into(),
subject_agent_id: ENGINEER.into(),
replacement_agent_id: "replacement-engineer".into(),
},
now(),
)
.expect("native root lineage authorization");
let replacement = broker_fixture::LiveProvider::launch_prepared(
&mut parent,
&endpoint,
&installed,
&root.join("provider"),
next,
&handle,
)
.expect("real replacement activation");
fs::write(
root.join(old_critic.result_artifact.as_ref().expect("Critic path")),
b"Deterministic rejected planning evidence.\n",
)
.expect("retained failed Critic");
engineer
.complete_child(old_critic.agent_id.as_str())
.expect("actual rejected Critic completion");
assert!(
engineer.complete().is_err(),
"malignant identity remains terminal"
);
engineer = replacement;
prefix = "replacement";
assert!(
!plan_fixture::invoke(&root, &["run", "orientation", "pre", RUN])
.status
.success(),
"replacement needs its own completed children"
);
assert_eq!(
fs::read(root.join(format!(".shepherd/runs/{RUN}/orientation-pre.json")))
.expect("retained old pre"),
original_pre
);
write_phase0(&root, prefix);
for role in [Role::Auditor, Role::Discovery] {
let child = engineer
.spawn(lineage_request(&root, role, prefix))
.expect("replacement's real child");
write_report(&root, &child);
engineer
.complete_child(child.agent_id.as_str())
.expect("replacement child completion");
}
let current_registry: serde_json::Value = serde_json::from_slice(
&fs::read(root.join(".shepherd/native-orientation-registry.json"))
.expect("ledger before retirement"),
)
.expect("ledger JSON");
let archive_path = root.join(format!(
".shepherd/runs/{RUN}/{}",
current_registry["runs"][RUN]["pre"]["input_archive"]["path"]
.as_str()
.expect("pre archive path")
));
let archived = fs::read(&archive_path).expect("immutable archive");
fs::write(&archive_path, b"changed archive\n").expect("archive negative fixture");
let denied = plan_fixture::invoke(&root, &["run", "orientation", "pre", RUN]);
assert!(
!denied.status.success(),
"tampered archive permitted native retirement"
);
assert!(
String::from_utf8_lossy(&denied.stderr)
.contains("immutable orientation input archive changed"),
"{}",
String::from_utf8_lossy(&denied.stderr)
);
fs::write(&archive_path, &archived).expect("restore exact archive");
let retained_registry: serde_json::Value = serde_json::from_slice(
&fs::read(root.join(".shepherd/native-orientation-registry.json"))
.expect("retained ledger"),
)
.expect("retained ledger JSON");
assert_eq!(retained_registry["runs"][RUN]["orientation_epoch"], 1);
assert!(
retained_registry["runs"][RUN]["pre_history"]
.as_array()
.expect("history")
.is_empty()
);
accepted(&root, &["run", "orientation", "pre", RUN, "--json"]);
let registry: serde_json::Value = serde_json::from_slice(
&fs::read(root.join(".shepherd/native-orientation-registry.json"))
.expect("native ledger"),
)
.expect("native ledger JSON");
let retired = ®istry["runs"][RUN]["pre_history"][0];
assert_eq!(retired["pre"]["epoch"], 1);
assert_eq!(retired["pre"]["engineer_nonce"], original.nonce);
assert_eq!(retired["replacement_agent_id"], "replacement-engineer");
assert_eq!(retired["post"]["post_sha256"], hash(&original_post));
let post_archive_ref = &retired["post"]["input_archive"];
let post_archive: serde_json::Value = serde_json::from_slice(
&fs::read(root.join(format!(
".shepherd/runs/{RUN}/{}",
post_archive_ref["path"].as_str().expect("post archive")
)))
.expect("post archive bytes"),
)
.expect("post archive JSON");
let post_bytes = fs::read(root.join(format!(
".shepherd/runs/{RUN}/{}",
post_archive["inputs"]["native/orientation-post.json"]["path"]
.as_str()
.expect("post bytes path")
)))
.expect("preserved post");
assert_eq!(post_bytes, original_post);
let archive_ref = &retired["pre"]["input_archive"];
let archive_bytes = fs::read(root.join(format!(
".shepherd/runs/{RUN}/{}",
archive_ref["path"].as_str().expect("archive index path")
)))
.expect("archive index bytes");
assert_eq!(hash(&archive_bytes), archive_ref["sha256"]);
let archive: serde_json::Value =
serde_json::from_slice(&archive_bytes).expect("archive index JSON");
let archives = archive["inputs"]
.as_object()
.expect("immutable source archives");
assert!(!archives.is_empty());
for (source, archive) in archives {
let bytes = fs::read(root.join(format!(
".shepherd/runs/{RUN}/{}",
archive["path"].as_str().expect("archive path")
)))
.expect("immutable bytes");
assert_eq!(hash(&bytes), archive["sha256"]);
if source == "native/orientation-pre.json" {
assert_eq!(bytes, original_pre);
}
}
assert_eq!(registry["runs"][RUN]["pre"]["epoch"], 2);
assert_eq!(
DispatchStore::new(root.join(".shepherd/runs"))
.load_for_run(&RunId::new(RUN).expect("run"), &original.agent_id)
.expect("old record")
.state,
DispatchState::Malignant
);
}
let baseline = plan_fixture::materialize(&root, RUN, &["lane-a"], 3);
let critic = engineer
.spawn(lineage_request(&root, Role::Critic, prefix))
.expect("Critic follows native pre");
write_report(&root, &critic);
engineer
.complete_child(critic.agent_id.as_str())
.expect("native Critic completion");
accepted(&root, &["run", "orientation", "post", RUN, "--json"]);
assert!(
Command::new("git")
.args([
"add",
"-f",
"--",
&format!(".shepherd/runs/{RUN}/plan-probes.json"),
&format!(".shepherd/runs/{RUN}/reports/{prefix}-critic.json")
])
.current_dir(&root)
.status()
.expect("stage reviewed planning")
.success()
);
assert!(
Command::new("git")
.args([
"-c",
"user.name=Shepherd Tests",
"-c",
"user.email=shepherd-tests@example.invalid",
"commit",
"--quiet",
"-m",
"reviewed planning descendant"
])
.current_dir(&root)
.status()
.expect("commit reviewed planning")
.success()
);
accepted(&root, &["run", "transition", RUN, "--to", "planned"]);
accepted(&root, &["sprint", "open", "--run", RUN]);
let state =
RunState::load(&root.join(format!(".shepherd/runs/{RUN}/run.json"))).expect("state");
assert_eq!(state.status, Vocabulary::Known(RunStatus::Executing));
assert_eq!(state.lanes.len(), 1);
assert_eq!(state.lanes[0].state, Vocabulary::Known(LaneStatus::Pending));
assert_ne!(state.extra["planning_execution_head"], baseline);
assert_eq!(
DispatchStore::new(root.join(".shepherd/runs"))
.load_for_run(&RunId::new(RUN).expect("run"), &engineer.record().agent_id)
.expect("persistent Engineer")
.state
.to_string(),
"active"
);
write_report(&root, engineer.record());
engineer
.complete()
.expect("native Engineer retirement after opening");
}