shepherd-cli 6.6.1

//! Real-process coverage for root-owned RED/GREEN, not a new Native gate API.
//!
//! The mailbox substitutes only for harness-native messages. The provider's
//! Native claim/activation, guard decisions, scoped edits, and Cargo processes
//! are real. These deterministic fixtures are not live LLM acceptance evidence.

#![cfg(unix)]

#[path = "support/broker.rs"]
mod broker_fixture;
#[path = "support/plan.rs"]
mod plan_fixture;
mod support;

use std::{
    fs,
    os::unix::fs::PermissionsExt,
    path::{Path, PathBuf},
    process::Command,
    thread,
    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};

use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use shepherd_cli::{
    BindRootDispatchRequest, CarrierAttachmentExpectationRequest, DispatchResolution,
    DispatchService, DispatchStore, NativeBroker, PreparePendingDispatchRequest,
    ResolveDispatchRequest,
    shepherd::{
        GuardEngine, GuardValue, Harness,
        compiler::content::embedded_guard_sources,
        dispatch::{
            DispatchRecord, DispatchState, PendingLaunchState, ProjectId, Role, RootSessionBinding,
            SessionId,
        },
        guard::{parse_predicate_toml, parse_role_markdown},
        plan::parse_plan,
    },
};

const RUN: &str = "v657";
const LANE: &str = "gate-lane";
const ROOT_SESSION: &str = "gate-root";
const SOURCE: &str = "src/plan-fixture-0.rs";
const TASK: &str = ".shepherd/runs/v657/lanes/gate-lane/plan.md";
const PLAN: &str = ".shepherd/runs/v657/plan.md";
const RED_SOURCE: &str = "fn fixture() {}\nconst _: () = { let _ = fixture; };\n\
    #[cfg(test)] mod tests {\n\
    #[test] fn gate_returns_two() {\n\
    assert_eq!(format!(\"{:?}\", super::fixture()), \"2\", \"ROOT_GATE_EXPECTED_TWO\");\n\
    }}\n";

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "lowercase")]
enum Phase {
    Red,
    Green,
}

impl Phase {
    fn name(self) -> &'static str {
        match self {
            Self::Red => "red",
            Self::Green => "green",
        }
    }
}

/// Test-only message shape. Native authority is always reloaded independently.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
struct GateMessage {
    phase: Phase,
    run: String,
    lane: String,
    node: String,
    dispatch_id: String,
    session_id: String,
    dispatch_revision: u64,
    root: RootSessionBinding,
    worktree: PathBuf,
    worktree_identity: String,
    baseline: String,
    task_path: String,
    task_sha256: String,
    plan_sha256: String,
    source_sha256: String,
    inputs_sha256: String,
    command: Vec<String>,
    provider_pid: u32,
}

#[derive(Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct GateReceipt {
    request: GateMessage,
    root: RootSessionBinding,
    root_pid: u32,
    exit_code: i32,
    stdout: Vec<u8>,
    stderr: Vec<u8>,
    stdout_sha256: String,
    stderr_sha256: String,
}

fn now() -> i64 {
    i64::try_from(
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("clock")
            .as_millis(),
    )
    .expect("milliseconds fit i64")
}

fn hash(bytes: &[u8]) -> String {
    Sha256::digest(bytes)
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect()
}

fn file_hash(root: &Path, path: &str) -> String {
    hash(&fs::read(root.join(path)).expect("measured fixture file"))
}

fn inputs_hash(root: &Path) -> String {
    let inputs = ["Cargo.lock", "Cargo.toml", SOURCE].map(|path| (path, file_hash(root, path)));
    hash(&serde_json::to_vec(&inputs).expect("sorted input inventory"))
}

fn git(root: &Path, args: &[&str]) -> String {
    let output = Command::new("git")
        .args(args)
        .current_dir(root)
        .output()
        .expect("fixture git");
    assert!(
        output.status.success(),
        "git {args:?}: {}",
        String::from_utf8_lossy(&output.stderr)
    );
    String::from_utf8(output.stdout)
        .expect("git output UTF-8")
        .trim()
        .to_owned()
}

fn committed_hash(root: &Path, commit: &str, path: &str) -> String {
    let output = Command::new("git")
        .args(["show", &format!("{commit}:{path}")])
        .current_dir(root)
        .output()
        .expect("exact baseline object");
    assert!(output.status.success(), "baseline fixture contains {path}");
    hash(&output.stdout)
}

fn atomic_json(path: &Path, value: &impl Serialize) {
    let temporary = path.with_extension("writing");
    fs::write(
        &temporary,
        serde_json::to_vec(value).expect("test message JSON"),
    )
    .expect("message bytes");
    fs::rename(temporary, path).expect("complete test message");
}

fn wait_json<T: serde::de::DeserializeOwned>(path: &Path) -> T {
    let deadline = Instant::now() + Duration::from_secs(45);
    while !path.is_file() {
        let error = path.parent().expect("mailbox").join("child-error.json");
        assert!(
            !error.exists(),
            "child failed: {}",
            fs::read_to_string(&error).unwrap_or_default()
        );
        assert!(
            Instant::now() < deadline,
            "timed out waiting for {}",
            path.display()
        );
        thread::sleep(Duration::from_millis(5));
    }
    serde_json::from_slice(&fs::read(path).expect("message bytes")).expect("complete message JSON")
}

fn service(root: &Path) -> DispatchService {
    let identity: Value = serde_json::from_slice(
        &fs::read(root.join(".shepherd/project.json")).expect("Native project file"),
    )
    .expect("project JSON");
    let installed = root.join("installed");
    DispatchService::with_project_root(
        DispatchStore::new(root.join(".shepherd/runs")),
        ProjectId::new(identity["id"].as_str().expect("project id")).expect("typed project id"),
        root,
    )
    .with_installed_package(&installed, installed.join(".shepherd-generated.json"))
}

fn child_identity(record: &DispatchRecord) -> ResolveDispatchRequest {
    ResolveDispatchRequest {
        schema: "shepherd.dispatch-request/1".into(),
        run: Some(record.run.to_string()),
        harness: record.harness,
        agent_id: Some(record.agent_id.to_string()),
        agent_type: Some(record.agent_type.to_string()),
        role_carrier: None,
        lane: record.lane.as_ref().map(ToString::to_string),
        session_id: record.session_id.to_string(),
        tool_call_id: None,
        tool_name: None,
        tool_input: None,
    }
}

/// This is the same Native-resolution plus embedded-guard sequence used by hooks.
fn authorize_tool(
    service: &DispatchService,
    request: ResolveDispatchRequest,
    at: i64,
) -> Result<DispatchResolution, String> {
    let resolution = service
        .resolve(request.clone(), at)
        .map_err(|error| error.to_string())?;
    let (predicates, roles) = embedded_guard_sources();
    let predicates = predicates
        .iter()
        .map(|(path, text)| parse_predicate_toml(path, text))
        .collect::<Result<Vec<_>, _>>()
        .expect("embedded predicates");
    let roles = roles
        .iter()
        .map(|(path, text)| parse_role_markdown(path, text))
        .collect::<Result<Vec<_>, _>>()
        .expect("embedded role facts");
    let engine = GuardEngine::new(predicates, roles).expect("canonical guard engine");
    let verdict = engine
        .evaluate(&GuardValue::from(json!({
            "role": resolution.role.as_str(), "dispatch": resolution,
            "tool_name": request.tool_name, "tool_input": request.tool_input,
        })))
        .expect("real guard evaluation");
    if verdict.decision.as_str() != "allow" {
        return Err(verdict
            .reason
            .unwrap_or_else(|| "canonical guard denied".into()));
    }
    Ok(resolution)
}

fn scoped_write(service: &DispatchService, root: &Path, record: &DispatchRecord, bytes: &str) {
    let mut request = child_identity(record);
    request.tool_name = Some("Write".into());
    request.tool_input = Some(json!({"file_path": SOURCE, "content": bytes}));
    let resolved =
        authorize_tool(service, request, now()).expect("live Coder owns this exact file");
    assert_eq!(resolved.path_in_write_scope, Some(true));
    assert_eq!(resolved.write_paths, [SOURCE]);
    fs::write(root.join(SOURCE), bytes).expect("actual child scoped edit");
}

fn gate_message(
    service: &DispatchService,
    root: &Path,
    record: &DispatchRecord,
    phase: Phase,
) -> GateMessage {
    let pending = service
        .store()
        .load_pending_for_agent(&record.run, &record.agent_id)
        .expect("Native task custody");
    GateMessage {
        phase,
        run: record.run.to_string(),
        lane: record.lane.as_ref().expect("Coder lane").to_string(),
        node: "node-0".into(),
        dispatch_id: record.agent_id.to_string(),
        session_id: record.session_id.to_string(),
        dispatch_revision: record.revision,
        root: service
            .store()
            .load_current_root_binding(&record.root_session_id)
            .expect("current root"),
        worktree: fs::canonicalize(root).expect("exact worktree"),
        worktree_identity: service
            .project_filesystem_id()
            .expect("Native worktree identity")
            .to_string(),
        baseline: pending.baseline_commit.to_string(),
        task_path: pending.task_path.to_string(),
        task_sha256: file_hash(root, pending.task_path.as_str()),
        plan_sha256: file_hash(root, PLAN),
        source_sha256: file_hash(root, SOURCE),
        inputs_sha256: inputs_hash(root),
        command: vec!["cargo".into(), "test".into()],
        provider_pid: std::process::id(),
    }
}

fn coder_flow(record: &DispatchRecord, root: &Path) {
    assert_eq!(record.state, DispatchState::Active);
    let service = service(root);
    let mailbox = root.join("host-messages");
    let mut shell = child_identity(record);
    shell.tool_name = Some("Bash".into());
    shell.tool_input = Some(json!({"command": "cargo test"}));
    let denied =
        authorize_tool(&service, shell, now()).expect_err("bounded Coder Bash fails before launch");
    assert!(denied.contains("opaque Bash effects"), "{denied}");
    atomic_json(
        &mailbox.join("shell-denied.json"),
        &json!({"reason": denied, "pid": std::process::id()}),
    );
    for path in ["src/not-owned.rs", "src/Plan-fixture-0.rs", "Cargo.toml"] {
        let mut outside = child_identity(record);
        outside.tool_name = Some("Write".into());
        outside.tool_input = Some(json!({"file_path": path, "content": "forbidden"}));
        assert!(
            authorize_tool(&service, outside, now()).is_err(),
            "read authority and case-insensitive filesystem aliases must not widen writes: {path}"
        );
    }
    assert_eq!(
        fs::read_to_string(root.join(SOURCE)).expect("original source"),
        "fn fixture() {}\n"
    );
    scoped_write(&service, root, record, RED_SOURCE);
    for phase in [Phase::Red, Phase::Green] {
        let request = gate_message(&service, root, record, phase);
        atomic_json(
            &mailbox.join(format!("{}-request.json", phase.name())),
            &request,
        );
        let receipt: GateReceipt =
            wait_json(&mailbox.join(format!("{}-receipt.json", phase.name())));
        assert_eq!(
            receipt.request, request,
            "receipt belongs to this exact task/revision/command"
        );
        assert_eq!(receipt.root, request.root, "same Native root authority");
        assert_eq!(
            receipt.root,
            service
                .store()
                .load_current_root_binding(&record.root_session_id)
                .expect("root authority still current")
        );
        assert_ne!(
            receipt.root_pid,
            std::process::id(),
            "Coder did not proxy a root process identity"
        );
        assert_eq!(receipt.stdout_sha256, hash(&receipt.stdout));
        assert_eq!(receipt.stderr_sha256, hash(&receipt.stderr));
        assert_eq!(
            inputs_hash(root),
            request.inputs_sha256,
            "no input drift while gate ran"
        );
        match phase {
            Phase::Red => {
                assert_eq!(
                    receipt.exit_code, 101,
                    "real assertion failure, not launch failure"
                );
                assert!(
                    String::from_utf8_lossy(&receipt.stdout).contains("ROOT_GATE_EXPECTED_TWO")
                );
                assert!(
                    String::from_utf8_lossy(&receipt.stdout)
                        .contains("tests::gate_returns_two ... FAILED")
                );
                assert_eq!(
                    fs::read_to_string(root.join(SOURCE)).expect("still unfixed"),
                    RED_SOURCE
                );
                // The production edit originates in the actual activated Coder PID,
                // and only after that process consumes the matching real RED.
                scoped_write(
                    &service,
                    root,
                    record,
                    &RED_SOURCE.replacen("fn fixture() {}", "fn fixture() -> u32 { 2 }", 1),
                );
            }
            Phase::Green => {
                assert_eq!(receipt.exit_code, 0);
                assert!(
                    String::from_utf8_lossy(&receipt.stdout)
                        .contains("tests::gate_returns_two ... ok")
                );
            }
        }
    }
    atomic_json(
        &mailbox.join("done.json"),
        &json!({"pid": std::process::id(), "state": "green",
            "result": "Real Native Coder received RED 101, edited its own source, then received GREEN 0.\n"}),
    );
}

#[test]
fn broker_fixture_child() {
    broker_fixture::child_main_with(|record| {
        if record.role != Role::Coder {
            return;
        }
        let path = std::env::var_os("SHEPHERD_TEST_BROKER_PROVIDER").expect("provider input");
        let input: Value = serde_json::from_slice(&fs::read(path).expect("provider bytes"))
            .expect("provider JSON");
        let installed = Path::new(input["installed"].as_str().expect("installed path"));
        let root = installed.parent().expect("fixture root");
        let result = std::panic::catch_unwind(|| coder_flow(record, root));
        if let Err(error) = result {
            atomic_json(
                &root.join("host-messages/child-error.json"),
                &"see provider process.log for the failing assertion",
            );
            std::panic::resume_unwind(error);
        }
    });
}

/// The root's independent review of a host message is workflow logic, not a
/// claim that Native has an arbitrary-command execution or sandbox API.
fn validate_message(
    service: &DispatchService,
    root: &Path,
    coder: &DispatchRecord,
    request: &GateMessage,
    at: i64,
) -> Result<(RootSessionBinding, String), String> {
    let root_request = ResolveDispatchRequest {
        schema: "shepherd.dispatch-request/1".into(),
        run: Some(request.run.clone()),
        harness: request.root.harness,
        agent_id: None,
        agent_type: None,
        role_carrier: None,
        lane: None,
        session_id: request.root.session_id.to_string(),
        tool_call_id: Some(format!("root-gate-{}", request.phase.name())),
        tool_name: Some("Bash".into()),
        tool_input: Some(json!({"command": "cargo test"})),
    };
    let resolved = authorize_tool(service, root_request, at)?;
    if resolved.role != Role::Shepherd
        || resolved.mode != Some(shepherd_cli::shepherd::dispatch::RootMode::Execution)
        || resolved.agent_id.is_some()
        || resolved.lane.is_some()
    {
        return Err("not the exact execution Shepherd root".into());
    }
    let binding = service
        .store()
        .load_current_root_binding(&coder.root_session_id)
        .map_err(|error| error.to_string())?;
    if binding != request.root || binding.session_id != resolved.session_id {
        return Err("stale or foreign root authority snapshot".into());
    }
    service
        .resolve(child_identity(coder), at)
        .map_err(|error| error.to_string())?;
    let active = service
        .store()
        .load_for_run(&coder.run, &coder.agent_id)
        .map_err(|error| error.to_string())?;
    let pending = service
        .store()
        .load_pending_for_agent(&coder.run, &coder.agent_id)
        .map_err(|error| error.to_string())?;
    if active != *coder
        || active.state != DispatchState::Active
        || active.role != Role::Coder
        || request.dispatch_id != active.agent_id.as_str()
        || request.session_id != active.session_id.as_str()
        || request.dispatch_revision != active.revision
        || request.run != active.run.as_str()
        || active.lane.as_ref().map(|lane| lane.as_str()) != Some(request.lane.as_str())
        || pending.launch_state != PendingLaunchState::Active
    {
        return Err("wrong, stale, or inactive Coder task identity".into());
    }
    if request.worktree != root
        || request.worktree_identity
            != service
                .project_filesystem_id()
                .map_err(|error| error.to_string())?
                .to_string()
        || request.baseline != pending.baseline_commit.as_str()
        || request.baseline != git(root, &["rev-parse", "HEAD"])
        || request.task_path != pending.task_path.as_str()
    {
        return Err("worktree, baseline, or retained task changed".into());
    }
    let task_digest: String = pending
        .task_sha256
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect();
    if request.task_sha256 != task_digest
        || request.task_sha256 != file_hash(root, pending.task_path.as_str())
        || request.plan_sha256 != file_hash(root, PLAN)
        || request.plan_sha256 != committed_hash(root, &request.baseline, PLAN)
        || request.source_sha256 != file_hash(root, SOURCE)
        || request.inputs_sha256 != inputs_hash(root)
        || ["Cargo.toml", "Cargo.lock"]
            .iter()
            .any(|path| file_hash(root, path) != committed_hash(root, &request.baseline, path))
    {
        return Err("task, plan, or command input bytes changed".into());
    }
    let plan = parse_plan(&fs::read_to_string(root.join(PLAN)).map_err(|error| error.to_string())?)
        .map_err(|error| error.to_string())?;
    let node = plan
        .manifest
        .nodes
        .iter()
        .find(|node| node.id == request.node)
        .ok_or("node is absent from current plan")?;
    let gate = match request.phase {
        Phase::Red => &node.red,
        Phase::Green => &node.green,
    };
    if plan.manifest.run != request.run
        || node.lane != request.lane
        || node.role != "coder"
        || node.owns != [SOURCE]
        || active.write_scope != [SOURCE]
        || request.task_path != TASK
        || gate.command != request.command
        || request.command != ["cargo", "test"]
        || gate.expects
            != match request.phase {
                Phase::Red => "failure",
                Phase::Green => "success",
            }
    {
        return Err("command or scope is not the exact planned Coder gate".into());
    }
    let lane = plan
        .manifest
        .capacity
        .cargo_targets
        .iter()
        .find(|lane| lane.lane == request.lane)
        .ok_or("lane has no Cargo target binding")?;
    Ok((binding, lane.value.clone()))
}

fn lane_command(root: &Path, lane: &str) -> Result<Command, String> {
    let target = PathBuf::from(
        std::env::var_os("CARGO_TARGET_DIR").ok_or("run through with-lane-target.sh")?,
    );
    if !target.is_absolute() {
        return Err("outer Cargo target must be absolute".into());
    }
    let wrapper = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../scripts/with-lane-target.sh");
    let mut command = Command::new(wrapper);
    command
        .arg(lane)
        .current_dir(root)
        .env("SHEPHERD_TARGET_ROOT", &target)
        .env("CARGO_NET_OFFLINE", "true");
    Ok(command)
}

fn execute_root_gate(
    service: &DispatchService,
    root: &Path,
    coder: &DispatchRecord,
    request: &GateMessage,
    at: i64,
    launches: &mut usize,
) -> Result<GateReceipt, String> {
    let (binding, lane) = validate_message(service, root, coder, request, at)?;
    let mut command = lane_command(root, &lane)?;
    *launches += 1;
    let output = command
        .args(&request.command)
        .output()
        .map_err(|error| error.to_string())?;
    let (after, _) = validate_message(service, root, coder, request, now())?;
    if after != binding {
        return Err("root authority changed while command ran".into());
    }
    Ok(GateReceipt {
        request: request.clone(),
        root: binding,
        root_pid: std::process::id(),
        exit_code: output.status.code().ok_or("gate terminated by signal")?,
        stdout_sha256: hash(&output.stdout),
        stderr_sha256: hash(&output.stderr),
        stdout: output.stdout,
        stderr: output.stderr,
    })
}

#[test]
fn live_coder_routes_real_red_green_to_current_root_without_shell_scope_widening() {
    let root = std::env::temp_dir().join(format!("shepherd-root-gate-{}", uuid::Uuid::now_v7()));
    fs::create_dir(&root).expect("isolated fixture");
    fs::set_permissions(&root, fs::Permissions::from_mode(0o700)).expect("private fixture");
    let root = fs::canonicalize(root).expect("canonical fixture root");
    eprintln!("root gate fixture: {}", root.display());
    fs::write(
        root.join(".gitignore"),
        "installed/\nprovider-fixtures/\nhost-messages/\nisolated-home/\n",
    )
    .expect("exclude test host transport");
    fs::write(root.join("Cargo.toml"), "[package]\nname = \"root-gate-fixture\"\nversion = \"0.0.0\"\nedition = \"2024\"\n[lib]\npath = \"src/plan-fixture-0.rs\"\n[workspace]\n")
        .expect("real dependency-free Cargo fixture");
    let lockfile = lane_command(&root, "plan-fixture-0")
        .expect("isolated Cargo lane")
        .args(["cargo", "generate-lockfile", "--offline"])
        .output()
        .expect("canonical Cargo lockfile");
    assert!(
        lockfile.status.success(),
        "{}",
        String::from_utf8_lossy(&lockfile.stderr)
    );
    fs::write(
        root.join("README.md"),
        "A real root-owned Cargo gate fixture.\n",
    )
    .expect("canonical uppercase README");
    fs::create_dir(root.join("host-messages")).expect("test host messages");
    git(&root, &["init", "--quiet"]);
    let baseline = plan_fixture::open_execution(&root, RUN, &[LANE], 4);
    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("canonical carrier compiler");
    assert!(
        compiled.status.success(),
        "{}",
        String::from_utf8_lossy(&compiled.stderr)
    );
    let service = service(&root);
    service
        .bind_root(
            BindRootDispatchRequest {
                schema: "shepherd.dispatch-request/1".into(),
                run: Some(RUN.into()),
                harness: Harness::Pi,
                session_id: ROOT_SESSION.into(),
                role_carrier: Role::Shepherd.carrier(),
                mode: shepherd_cli::shepherd::dispatch::RootMode::Execution,
                lease_ms: 600_000,
            },
            now(),
        )
        .expect("exact root bound natively");
    let mut root_write = ResolveDispatchRequest {
        schema: "shepherd.dispatch-request/1".into(),
        run: Some(RUN.into()),
        harness: Harness::Pi,
        agent_id: None,
        agent_type: None,
        role_carrier: None,
        lane: None,
        session_id: ROOT_SESSION.into(),
        tool_call_id: Some("root-structured-write".into()),
        tool_name: Some("Write".into()),
        tool_input: Some(
            json!({"file_path": SOURCE, "content": "root must not implement the fix"}),
        ),
    };
    assert!(
        authorize_tool(&service, root_write.clone(), now()).is_err(),
        "root gate execution must not grant root structured production writes"
    );
    root_write.tool_input =
        Some(json!({"file_path": "README.md", "content": "root-owned Markdown"}));
    let markdown =
        authorize_tool(&service, root_write, now()).expect("canonical mixed-case Markdown path");
    assert_eq!(
        markdown.write_paths,
        ["README.md"],
        "Native preserves path case"
    );
    assert_eq!(markdown.path_in_write_scope, Some(true));
    let endpoint = fs::canonicalize("/tmp")
        .expect("short socket root")
        .join(format!("sg-{}-{}", std::process::id(), now()))
        .join("broker.sock");
    let broker = NativeBroker::start(service.clone(), &endpoint).expect("real Native broker");
    let mut parent = broker.connect().expect("root native peer");
    parent
        .register_parent(
            Harness::Pi,
            Role::Shepherd,
            SessionId::new(ROOT_SESSION).expect("root id"),
            SessionId::new(ROOT_SESSION).expect("root id"),
            None,
        )
        .expect("root registered by actual PID");
    let request = |role: Role| PreparePendingDispatchRequest {
        schema: "shepherd.pending-dispatch-request/2".into(),
        run: Some(RUN.into()),
        role: role.to_string(),
        work_kind: if role == Role::Coder {
            "production-code"
        } else {
            "coordination"
        }
        .into(),
        lane: Some(LANE.into()),
        parent_dispatch_id: (role == Role::Coder).then(|| "gate-conductor".into()),
        replaces_agent_id: None,
        baseline: baseline.clone(),
        read_scope: vec![
            TASK.into(),
            PLAN.into(),
            SOURCE.into(),
            "Cargo.toml".into(),
            "Cargo.lock".into(),
            "README.md".into(),
        ],
        write_scope: if role == Role::Coder {
            vec![SOURCE.into()]
        } else {
            vec![format!(
                ".shepherd/runs/{RUN}/lanes/{LANE}/reports/conductor.md"
            )]
        },
        result_artifact: format!(".shepherd/runs/{RUN}/lanes/{LANE}/reports/{role}.md"),
        review_artifact: format!(".shepherd/runs/{RUN}/lanes/{LANE}/reviews/{role}.md"),
        task_file: TASK.into(),
        child_session_id: format!("gate-{role}-session"),
        lease_ms: 600_000,
        expected_attachment: CarrierAttachmentExpectationRequest {
            target: Harness::Pi,
            role: role.to_string(),
            agent_id: format!("gate-{role}"),
            attachment_kind: "pi-skill-path".into(),
        },
    };
    let mut conductor = broker_fixture::LiveProvider::launch(
        &mut parent,
        &endpoint,
        &installed,
        &root.join("provider-fixtures"),
        request(Role::Conductor),
    )
    .expect("actual Conductor claim/activation");
    let coder = conductor
        .spawn(request(Role::Coder))
        .expect("actual child Coder claim/activation");
    assert_eq!(
        coder
            .parent_agent_id
            .as_ref()
            .expect("live Conductor")
            .as_str(),
        "gate-conductor"
    );
    let mailbox = root.join("host-messages");
    let denied: Value = wait_json(&mailbox.join("shell-denied.json"));
    assert!(
        denied["reason"]
            .as_str()
            .expect("denial reason")
            .contains("opaque Bash effects")
    );
    assert_ne!(denied["pid"], std::process::id());
    let red: GateMessage = wait_json(&mailbox.join("red-request.json"));
    assert_eq!(
        fs::read_to_string(root.join(SOURCE)).expect("tests precede fix"),
        RED_SOURCE
    );
    let mut launches = 0;
    for label in [
        "wrong-root",
        "wrong-harness",
        "wrong-run",
        "wrong-session",
        "wrong-dispatch",
        "wrong-lane",
        "stale-root",
        "stale-revision",
        "wrong-worktree",
        "wrong-worktree-identity",
        "wrong-baseline",
        "wrong-task",
        "wrong-task-digest",
        "stale-plan",
        "stale-source",
        "stale-inputs",
        "wrong-node",
        "wrong-command",
        "expired-root",
    ] {
        let mut invalid = red.clone();
        let mut at = now();
        match label {
            "wrong-root" => {
                invalid.root.session_id = SessionId::new("never-bound").expect("unbound identity")
            }
            "wrong-harness" => invalid.root.harness = Harness::Codex,
            "wrong-run" => invalid.run = "v999".into(),
            "wrong-session" => invalid.session_id = "other-coder".into(),
            "wrong-dispatch" => invalid.dispatch_id = "other-coder".into(),
            "wrong-lane" => invalid.lane = "other-lane".into(),
            "stale-root" => invalid.root.bound_at += 1,
            "stale-revision" => invalid.dispatch_revision += 1,
            "wrong-worktree" => invalid.worktree = installed.clone(),
            "wrong-worktree-identity" => invalid.worktree_identity = "0".repeat(64),
            "wrong-baseline" => invalid.baseline = "f".repeat(40),
            "wrong-task" => invalid.task_path = PLAN.into(),
            "wrong-task-digest" => invalid.task_sha256 = "0".repeat(64),
            "stale-plan" => invalid.plan_sha256 = "0".repeat(64),
            "stale-source" => invalid.source_sha256 = "0".repeat(64),
            "stale-inputs" => invalid.inputs_sha256 = "0".repeat(64),
            "wrong-node" => invalid.node = "absent-node".into(),
            "wrong-command" => invalid.command.push("--lib".into()),
            "expired-root" => at = invalid.root.expires_at,
            _ => unreachable!(),
        }
        let error = execute_root_gate(&service, &root, &coder, &invalid, at, &mut launches)
            .expect_err("foreign or stale message cannot launch a command");
        eprintln!("root-gate negative {label}: {error}");
        assert_eq!(launches, 0, "{label} was denied before process creation");
        assert_eq!(
            fs::read_to_string(root.join(SOURCE)).expect("no denied mutation"),
            RED_SOURCE
        );
    }
    let red_receipt = execute_root_gate(&service, &root, &coder, &red, now(), &mut launches)
        .expect("exact authorized root executes actual RED");
    assert_eq!(red_receipt.exit_code, 101);
    eprintln!(
        "real RED exit {}\n{}",
        red_receipt.exit_code,
        String::from_utf8_lossy(&red_receipt.stdout)
    );
    assert_eq!(
        fs::read_to_string(root.join(SOURCE)).expect("no production edit before RED receipt"),
        RED_SOURCE
    );
    atomic_json(&mailbox.join("red-receipt.json"), &red_receipt);
    let green: GateMessage = wait_json(&mailbox.join("green-request.json"));
    assert_eq!(green.dispatch_revision, red.dispatch_revision);
    assert_eq!(green.baseline, red.baseline);
    assert_ne!(green.source_sha256, red.source_sha256);
    let green_receipt = execute_root_gate(&service, &root, &coder, &green, now(), &mut launches)
        .expect("exact authorized root executes actual GREEN");
    assert_eq!(
        green_receipt.exit_code,
        0,
        "{}",
        String::from_utf8_lossy(&green_receipt.stderr)
    );
    eprintln!(
        "real GREEN exit {}\n{}",
        green_receipt.exit_code,
        String::from_utf8_lossy(&green_receipt.stdout)
    );
    atomic_json(&mailbox.join("green-receipt.json"), &green_receipt);
    let done: Value = wait_json(&mailbox.join("done.json"));
    assert_eq!(done["pid"], red.provider_pid);
    assert_eq!(launches, 2);
    assert!(
        !root.join("target").exists(),
        "gate must use its explicit lane target"
    );
    assert!(!root.join("src/not-owned.rs").exists());
    assert_eq!(
        fs::read_to_string(root.join(SOURCE)).expect("Coder source"),
        RED_SOURCE.replacen("fn fixture() {}", "fn fixture() -> u32 { 2 }", 1)
    );
    // The test host persists the child's native result message at Native's
    // declared result path. This is not a second Coder file-write capability.
    let result = root.join(
        coder
            .result_artifact
            .as_deref()
            .expect("Native result path"),
    );
    fs::create_dir_all(result.parent().expect("result directory")).expect("result directory");
    fs::write(
        result,
        done["result"]
            .as_str()
            .expect("actual child result message"),
    )
    .expect("host retained child result");
    assert_eq!(
        conductor
            .complete_child(coder.agent_id.as_str())
            .expect("Coder native completion")
            .state,
        DispatchState::Stopped
    );
    assert!(
        execute_root_gate(&service, &root, &coder, &green, now(), &mut launches).is_err(),
        "a stopped Coder cannot replay a gate request"
    );
    assert_eq!(launches, 2);
    let result = root.join(
        conductor
            .record()
            .result_artifact
            .as_deref()
            .expect("Conductor result"),
    );
    fs::write(
        result,
        "Real Coder RED/GREEN completed with exact root gate receipts.\n",
    )
    .expect("Conductor result");
    conductor.complete().expect("Conductor native completion");
    drop(conductor);
    drop(parent);
    drop(broker);
    support::remove_dir_all(&root);
    support::remove_dir_all(endpoint.parent().expect("unique socket directory"));
}