shepherd-cli 6.6.1

//! Native provider process transport over the OS-authenticated launch broker.
//!
//! This command boundary is deliberately smaller than the ordinary dispatch
//! CLI.  A parent submits one typed launch plan over stdin; the native process
//! starts the broker, prepares the durable launch, starts the absolute child
//! executable, registers its real PID, and owns cleanup.  Child hooks submit
//! only an event and native-issued correlation fields.  They never carry a
//! pending path, token, descriptor, status, or result path.

use std::{
    collections::BTreeSet,
    fs::{self, OpenOptions},
    io::{self, Write},
    path::{Component, Path, PathBuf},
    process::{Command, Stdio},
    thread,
    time::{Duration, Instant},
};

use serde::{Deserialize, Serialize};
use shepherd::{
    Harness,
    dispatch::{AgentType, DispatchRecord, Role, SessionId},
};

use crate::{
    BrokerClient, BrokerLaunchId, DispatchService, DispatchStore, ExecutionContext, NativeBroker,
    PreparePendingDispatchRequest, interface::CliError,
};

#[cfg(unix)]
use std::{io::Read, os::unix::fs::PermissionsExt};

const LAUNCH_SCHEMA: &str = "shepherd.native-launch-request/1";
const CHILD_EVENT_SCHEMA: &str = "shepherd.native-child-event/1";
const CHILD_RESPONSE_SCHEMA: &str = "shepherd.native-child-response/1";
const MAX_INPUT_BYTES: usize = 1_048_576;
const MAX_ARGS: usize = 256;
const MAX_ENV: usize = 128;
const MAX_TIMEOUT_MS: u64 = 86_400_000;
const MAX_AUTH_BYTES: usize = 256 * 1024;

#[derive(Clone, Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct NativeLaunchRequest {
    pub schema: String,
    pub harness: Harness,
    pub parent_role: String,
    pub parent_session_id: String,
    pub root_session_id: String,
    pub parent_dispatch_id: Option<String>,
    pub prepare: PreparePendingDispatchRequest,
    pub executable: String,
    #[serde(default)]
    pub args: Vec<String>,
    #[serde(default)]
    pub env: Vec<LaunchEnvironmentEntry>,
    #[serde(default)]
    pub auth_snapshot: Option<AuthSnapshot>,
    pub timeout_ms: u64,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct LaunchEnvironmentEntry {
    pub name: String,
    pub value: String,
}

#[derive(Clone, Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct AuthSnapshot {
    pub source: String,
    pub destination: String,
}

#[derive(
    Clone,
    Copy,
    Debug,
    Deserialize,
    Eq,
    PartialEq,
    Serialize,
    strum::AsRefStr,
    strum::Display,
    strum::EnumCount,
    strum::EnumIs,
    strum::EnumString,
    strum::IntoStaticStr,
    strum::VariantNames,
)]
#[serde(rename_all = "kebab-case")]
#[strum(ascii_case_insensitive, serialize_all = "snake_case")]
pub(crate) enum ChildEvent {
    Start,
    Resume,
    Stop,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct NativeChildEventRequest {
    pub schema: String,
    pub harness: Harness,
    pub event: ChildEvent,
    pub session_id: String,
    pub agent_id: String,
    pub agent_type: String,
}

#[derive(Clone, Debug, Serialize)]
#[serde(deny_unknown_fields)]
struct NativeChildEventResponse {
    schema: String,
    harness: Harness,
    event: ChildEvent,
    record: DispatchRecord,
}

pub(crate) fn run_launch(mut context: ExecutionContext) -> Result<(), CliError> {
    let request: NativeLaunchRequest = read_json(&mut context)?;
    let child_role = validate_launch_request(&request, &context.workspace_root)?;
    let project_id = super::dispatch::read_project_id(&context.project_id_path)?;
    let service = DispatchService::with_context(
        DispatchStore::new(&context.runs_root),
        project_id,
        &context.workspace_root,
        &context.registry_path,
    );
    let endpoint_dir = private_endpoint_dir()?;
    let endpoint = endpoint_dir.join("broker.sock");
    let broker = NativeBroker::start(service, &endpoint)
        .map_err(|error| CliError::message(format!("native broker start failed: {error}")))?;
    let mut parent = broker.connect().map_err(|error| {
        CliError::message(format!("native broker parent connection failed: {error}"))
    })?;
    let parent_role = parse_role(&request.parent_role)?;
    let parent_session = SessionId::new(request.parent_session_id.clone())
        .map_err(|error| CliError::message(error.to_string()))?;
    let root_session = SessionId::new(request.root_session_id.clone())
        .map_err(|error| CliError::message(error.to_string()))?;
    let dispatch_id = request
        .parent_dispatch_id
        .as_deref()
        .map(crate::shepherd::dispatch::DispatchId::new)
        .transpose()
        .map_err(|error| CliError::message(error.to_string()))?;
    parent
        .register_parent(
            request.harness,
            parent_role,
            parent_session,
            root_session,
            dispatch_id,
        )
        .map_err(|error| {
            CliError::message(format!("native broker parent registration failed: {error}"))
        })?;
    let launch = parent
        .prepare(request.prepare.clone())
        .map_err(|error| CliError::message(format!("native broker prepare failed: {error}")))?;

    let home = endpoint_dir.join("home");
    let child = match spawn_provider(
        &request,
        child_role,
        &context.workspace_root,
        &home,
        &endpoint,
        &launch,
    ) {
        Ok(child) => child,
        Err(error) => {
            let _ = parent.cleanup(
                &launch,
                shepherd::dispatch::PendingLaunchState::LaunchFailed,
            );
            return Err(error);
        }
    };
    if let Err(error) = parent.register_child(&launch, child.id()) {
        let mut child = child;
        let _ = child.kill();
        let _ = child.wait();
        let _ = parent.cleanup(
            &launch,
            shepherd::dispatch::PendingLaunchState::LaunchFailed,
        );
        return Err(CliError::message(format!(
            "native broker child registration failed: {error}"
        )));
    }
    wait_for_provider(child, &mut parent, &launch, request.timeout_ms)
}

pub(crate) fn run_child(mut context: ExecutionContext) -> Result<(), CliError> {
    let request: NativeChildEventRequest = read_json(&mut context)?;
    validate_child_event(&request)?;
    let endpoint = required_native_env("SHEPHERD_NATIVE_BROKER_ENDPOINT")?;
    let launch_id =
        BrokerLaunchId::from_opaque(&required_native_env("SHEPHERD_NATIVE_BROKER_LAUNCH_ID")?)
            .map_err(|error| {
                CliError::message(format!("native broker launch identity rejected: {error}"))
            })?;
    for (name, expected) in [
        ("SHEPHERD_NATIVE_BROKER_SESSION_ID", &request.session_id),
        ("SHEPHERD_NATIVE_BROKER_AGENT_ID", &request.agent_id),
        ("SHEPHERD_NATIVE_BROKER_AGENT_TYPE", &request.agent_type),
    ] {
        if required_native_env(name)? != *expected {
            return Err(CliError::message(format!(
                "native broker {name} does not match the hook event"
            )));
        }
    }
    let (child, trusted_harness) = match request.event {
        ChildEvent::Start | ChildEvent::Resume => {
            let mut child =
                BrokerClient::connect_child_by_id(&endpoint, launch_id).map_err(|error| {
                    CliError::message(format!("native broker child connection failed: {error}"))
                })?;
            let trusted_harness = child
                .expected_attachment()
                .map_err(|error| {
                    CliError::message(format!("native broker attachment exchange failed: {error}"))
                })?
                .target;
            if request.harness != trusted_harness {
                return Err(CliError::message(
                    "native broker child harness does not match the prepared attachment",
                ));
            }
            let attestation = child.loaded_carrier_attestation().map_err(|error| {
                CliError::message(format!("native broker attachment exchange failed: {error}"))
            })?;
            child
                .claim(
                    request.agent_id.clone(),
                    request.session_id.clone(),
                    request.agent_type.clone(),
                    attestation.clone(),
                )
                .map_err(|error| {
                    CliError::message(format!("native broker claim failed: {error}"))
                })?;
            let record = child
                .activate(
                    request.agent_id.clone(),
                    request.session_id.clone(),
                    request.agent_type.clone(),
                    attestation,
                )
                .map_err(|error| {
                    CliError::message(format!("native broker activation failed: {error}"))
                })?;
            (record, trusted_harness)
        }
        ChildEvent::Stop => {
            let mut child =
                BrokerClient::connect_child_event_by_id(&endpoint, launch_id).map_err(|error| {
                    CliError::message(format!("native broker terminal connection failed: {error}"))
                })?;
            let trusted_harness = child
                .expected_attachment()
                .map_err(|error| {
                    CliError::message(format!("native broker attachment exchange failed: {error}"))
                })?
                .target;
            if request.harness != trusted_harness {
                return Err(CliError::message(
                    "native broker child harness does not match the active attachment",
                ));
            }
            let record = child
                .complete(
                    request.agent_id.clone(),
                    request.session_id.clone(),
                    request.agent_type.clone(),
                )
                .map_err(|error| {
                    CliError::message(format!("native broker completion failed: {error}"))
                })?;
            (record, trusted_harness)
        }
    };
    let response = NativeChildEventResponse {
        schema: CHILD_RESPONSE_SCHEMA.into(),
        harness: trusted_harness,
        event: request.event,
        record: child,
    };
    write_json(&mut context, &response)
}

fn parse_role(value: &str) -> Result<Role, CliError> {
    Role::from_carrier(value)
        .or_else(|_| Role::from_name(value))
        .map_err(|error| CliError::message(error.to_string()))
}

fn validate_launch_request(
    request: &NativeLaunchRequest,
    project_root: &Path,
) -> Result<Role, CliError> {
    if request.schema != LAUNCH_SCHEMA {
        return Err(CliError::message(
            "native launch request schema is unsupported",
        ));
    }
    if request.prepare.schema != "shepherd.pending-dispatch-request/2" {
        return Err(CliError::message(
            "native launch prepare schema is unsupported",
        ));
    }
    let executable = Path::new(&request.executable);
    if !executable.is_absolute() {
        return Err(CliError::message(
            "provider executable must be an absolute pinned path",
        ));
    }
    let metadata = fs::symlink_metadata(executable).map_err(|error| {
        CliError::message(format!("cannot inspect provider executable: {error}"))
    })?;
    if metadata.file_type().is_symlink() || !metadata.is_file() {
        return Err(CliError::message(
            "provider executable must be a regular non-symlink file",
        ));
    }
    #[cfg(unix)]
    if metadata.permissions().mode() & 0o111 == 0 {
        return Err(CliError::message("provider executable is not executable"));
    }
    if request.args.len() > MAX_ARGS
        || request
            .args
            .iter()
            .any(|arg| arg.len() > 32 * 1024 || arg.chars().any(char::is_control))
    {
        return Err(CliError::message(
            "provider arguments exceed the native launch contract",
        ));
    }
    if request.env.len() > MAX_ENV {
        return Err(CliError::message(
            "provider environment exceeds the explicit allowlist",
        ));
    }
    let mut names = BTreeSet::new();
    for entry in &request.env {
        validate_environment_entry(entry)?;
        if !names.insert(entry.name.clone()) {
            return Err(CliError::message(
                "provider environment contains a duplicate name",
            ));
        }
    }
    if request.timeout_ms == 0 || request.timeout_ms > MAX_TIMEOUT_MS {
        return Err(CliError::message(
            "provider timeout is outside the native bound",
        ));
    }
    let project_root = fs::canonicalize(project_root).map_err(|error| {
        CliError::message(format!("cannot resolve native project root: {error}"))
    })?;
    if !project_root.is_dir() {
        return Err(CliError::message("native project root is not a directory"));
    }
    let parent_session = SessionId::new(request.parent_session_id.clone())
        .map_err(|error| CliError::message(error.to_string()))?;
    let root_session = SessionId::new(request.root_session_id.clone())
        .map_err(|error| CliError::message(error.to_string()))?;
    if parent_session == root_session && request.parent_dispatch_id.is_some() {
        return Err(CliError::message(
            "a root session cannot carry parent dispatch ancestry",
        ));
    }
    let _role = parse_role(&request.parent_role)?;
    if request.prepare.expected_attachment.target != request.harness {
        return Err(CliError::message(
            "native launch harness does not match the trusted attachment target",
        ));
    }
    if request.prepare.parent_dispatch_id != request.parent_dispatch_id {
        return Err(CliError::message(
            "native launch parent ancestry does not match prepare",
        ));
    }
    let child_role = parse_role(&request.prepare.role)?;
    if parse_role(&request.prepare.expected_attachment.role)? != child_role {
        return Err(CliError::message("native launch role contract is invalid"));
    }
    validate_native_path(&request.executable)?;
    if let Some(auth) = &request.auth_snapshot {
        validate_auth_snapshot(auth, &project_root)?;
    }
    Ok(child_role)
}

fn validate_child_event(request: &NativeChildEventRequest) -> Result<(), CliError> {
    if request.schema != CHILD_EVENT_SCHEMA {
        return Err(CliError::message(
            "native child event schema is unsupported",
        ));
    }
    SessionId::new(request.session_id.clone())
        .map_err(|error| CliError::message(error.to_string()))?;
    crate::shepherd::dispatch::AgentId::new(request.agent_id.clone())
        .map_err(|error| CliError::message(error.to_string()))?;
    AgentType::new(request.agent_type.clone())
        .map_err(|error| CliError::message(error.to_string()))?;
    Ok(())
}

fn validate_environment_entry(entry: &LaunchEnvironmentEntry) -> Result<(), CliError> {
    if entry.name.is_empty()
        || entry.name.len() > 128
        || !entry
            .name
            .bytes()
            .all(|byte| byte.is_ascii_uppercase() || byte.is_ascii_digit() || byte == b'_')
        || entry.name.starts_with("SHEPHERD_NATIVE_BROKER")
        || entry.name.starts_with("PI_SUBAGENT_")
        || entry.name.contains("SESSION")
        || entry.name.contains("ROLE")
        || entry.name.contains("AGENT")
        || entry.name.contains("PROJECT")
    {
        return Err(CliError::message(
            "provider environment contains an ambient identity variable",
        ));
    }
    if entry.value.len() > 64 * 1024 || entry.value.chars().any(char::is_control) {
        return Err(CliError::message("provider environment value is invalid"));
    }
    Ok(())
}

fn validate_native_path(path: &str) -> Result<(), CliError> {
    let path = Path::new(path);
    if path
        .components()
        .any(|component| matches!(component, Component::ParentDir))
    {
        return Err(CliError::message(
            "native launch paths cannot contain parent traversal",
        ));
    }
    Ok(())
}

fn validate_auth_snapshot(auth: &AuthSnapshot, _project_root: &Path) -> Result<(), CliError> {
    if !Path::new(&auth.source).is_absolute()
        || auth.destination.is_empty()
        || Path::new(&auth.destination).is_absolute()
        || Path::new(&auth.destination)
            .components()
            .any(|component| !matches!(component, Component::Normal(_)))
    {
        return Err(CliError::message(
            "auth snapshot path is not a bounded relative destination",
        ));
    }
    let _ = read_nofollow(Path::new(&auth.source), MAX_AUTH_BYTES)?;
    Ok(())
}

fn spawn_provider(
    request: &NativeLaunchRequest,
    child_role: Role,
    project_root: &Path,
    home: &Path,
    endpoint: &Path,
    launch: &crate::LaunchHandle,
) -> Result<std::process::Child, CliError> {
    fs::create_dir_all(home).map_err(|error| {
        CliError::message(format!("cannot create isolated provider HOME: {error}"))
    })?;
    set_private_directory(home)?;
    for directory in [
        home.join(".config"),
        home.join(".state"),
        home.join(".cache"),
    ] {
        fs::create_dir_all(&directory).map_err(|error| {
            CliError::message(format!("cannot create isolated provider state: {error}"))
        })?;
        set_private_directory(&directory)?;
    }
    if let Some(auth) = &request.auth_snapshot {
        let bytes = read_nofollow(Path::new(&auth.source), MAX_AUTH_BYTES)?;
        let destination = home.join(&auth.destination);
        if let Some(parent) = destination.parent() {
            fs::create_dir_all(parent).map_err(|error| {
                CliError::message(format!("cannot create isolated auth directory: {error}"))
            })?;
            set_private_directory(parent)?;
        }
        publish_private_file(&destination, &bytes)?;
    }
    let mut command = Command::new(&request.executable);
    command.args(&request.args);
    command.current_dir(project_root);
    command.env_clear();
    for entry in &request.env {
        command.env(&entry.name, &entry.value);
    }
    command.env("HOME", home);
    command.env("XDG_CONFIG_HOME", home.join(".config"));
    command.env("XDG_STATE_HOME", home.join(".state"));
    command.env("XDG_CACHE_HOME", home.join(".cache"));
    match request.harness {
        Harness::ClaudeCode => command.env("CLAUDE_CONFIG_DIR", home.join(".config/claude")),
        Harness::Codex => command.env("CODEX_HOME", home.join(".config/codex")),
        Harness::Pi => command.env("PI_CONFIG_DIR", home.join(".config/pi")),
        _ => {
            return Err(CliError::message(
                "native broker does not launch this harness",
            ));
        }
    };
    command.env("SHEPHERD_NATIVE_BROKER_ENDPOINT", endpoint);
    command.env(
        "SHEPHERD_NATIVE_BROKER_LAUNCH_ID",
        launch.launch_id().opaque_string(),
    );
    command.env(
        "SHEPHERD_NATIVE_BROKER_SESSION_ID",
        &request.prepare.child_session_id,
    );
    command.env(
        "SHEPHERD_NATIVE_BROKER_AGENT_ID",
        &request.prepare.expected_attachment.agent_id,
    );
    command.env(
        "SHEPHERD_NATIVE_BROKER_AGENT_TYPE",
        native_agent_type(request.harness, child_role),
    );
    command.stdin(Stdio::inherit());
    // The broker launch command has a strict empty-stdout protocol. Provider
    // output must never be allowed to splice into its JSON-free control reply.
    command.stdout(Stdio::null());
    command.stderr(Stdio::inherit());
    command
        .spawn()
        .map_err(|error| CliError::message(format!("provider process spawn failed: {error}")))
}

fn wait_for_provider(
    mut child: std::process::Child,
    parent: &mut BrokerClient,
    launch: &crate::LaunchHandle,
    timeout_ms: u64,
) -> Result<(), CliError> {
    let deadline = Instant::now() + Duration::from_millis(timeout_ms);
    loop {
        match child.try_wait() {
            Ok(Some(status)) => {
                if !status.success() {
                    let _ = parent
                        .cleanup(launch, shepherd::dispatch::PendingLaunchState::LaunchFailed);
                    return Err(CliError::message(format!(
                        "provider process exited with {status}"
                    )));
                }
                let _ = parent.cleanup(launch, shepherd::dispatch::PendingLaunchState::Canceled);
                return Ok(());
            }
            Ok(None) if Instant::now() >= deadline => {
                let _ = child.kill();
                let _ = child.wait();
                let _ = parent.cleanup(launch, shepherd::dispatch::PendingLaunchState::Expired);
                return Err(CliError::message(
                    "provider process exceeded its native launch lease",
                ));
            }
            Ok(None) => thread::sleep(Duration::from_millis(10)),
            Err(error) => {
                let _ =
                    parent.cleanup(launch, shepherd::dispatch::PendingLaunchState::LaunchFailed);
                return Err(CliError::message(format!(
                    "provider process wait failed: {error}"
                )));
            }
        }
    }
}

fn native_agent_type(harness: Harness, role: Role) -> String {
    match harness {
        Harness::Codex => {
            if role.write_eligible() {
                "worker".into()
            } else {
                "explorer".into()
            }
        }
        Harness::Pi => format!("pi-subagents:{}", role.as_str()),
        _ => role.as_str().to_owned(),
    }
}

fn private_endpoint_dir() -> Result<PathBuf, CliError> {
    let root = endpoint_root(&std::env::temp_dir());
    for _ in 0..16 {
        let suffix = BrokerLaunchId::fresh()
            .map_err(|error| CliError::message(error.to_string()))?
            .opaque_string();
        let path = root.join(format!("shepherd-broker-{suffix}"));
        match fs::create_dir(&path) {
            Ok(()) => {
                set_private_directory(&path)?;
                return Ok(path);
            }
            Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
            Err(error) => {
                return Err(CliError::message(format!(
                    "cannot create private broker directory: {error}"
                )));
            }
        }
    }
    Err(CliError::message(
        "cannot allocate a private native broker directory",
    ))
}

fn endpoint_root(preferred: &Path) -> PathBuf {
    #[cfg(unix)]
    {
        use std::os::unix::ffi::OsStrExt;
        let longest = preferred
            .join(format!("shepherd-broker-{}", "0".repeat(64)))
            .join("broker.sock");
        if longest.as_os_str().as_bytes().len() >= 100 {
            return PathBuf::from("/tmp");
        }
    }
    preferred.to_path_buf()
}

fn set_private_directory(_path: &Path) -> Result<(), CliError> {
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(_path, fs::Permissions::from_mode(0o700)).map_err(|error| {
            CliError::message(format!("cannot restrict native broker path: {error}"))
        })?;
    }
    Ok(())
}

fn publish_private_file(path: &Path, bytes: &[u8]) -> Result<(), CliError> {
    let result = (|| {
        let mut options = OpenOptions::new();
        options.write(true).create_new(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            options.mode(0o600);
        }
        let mut file = options
            .open(path)
            .map_err(|error| CliError::message(format!("cannot create auth snapshot: {error}")))?;
        file.write_all(bytes)
            .map_err(|error| CliError::message(format!("cannot publish auth snapshot: {error}")))?;
        file.sync_all()
            .map_err(|error| CliError::message(format!("cannot sync auth snapshot: {error}")))?;
        #[cfg(unix)]
        if let Some(parent) = path.parent() {
            fs::File::open(parent)
                .and_then(|directory| directory.sync_all())
                .map_err(|error| {
                    CliError::message(format!("cannot sync auth snapshot directory: {error}"))
                })?;
        }
        Ok(())
    })();
    if result.is_err() {
        let _ = fs::remove_file(path);
    }
    result
}

fn read_nofollow(path: &Path, limit: usize) -> Result<Vec<u8>, CliError> {
    #[cfg(unix)]
    {
        use rustix::fs::{FileType, Mode, OFlags, fstat, open};
        let descriptor = open(
            path,
            OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW | OFlags::NONBLOCK,
            Mode::empty(),
        )
        .map_err(|error| CliError::message(format!("cannot open nofollow file: {error}")))?;
        let stat = fstat(&descriptor)
            .map_err(|error| CliError::message(format!("cannot inspect nofollow file: {error}")))?;
        if !FileType::from_raw_mode(stat.st_mode).is_file() || stat.st_nlink != 1 {
            return Err(CliError::message(
                "nofollow file must be a regular file with one link",
            ));
        }
        let file = fs::File::from(descriptor);
        let mut bytes = Vec::new();
        file.take(u64::try_from(limit.saturating_add(1)).expect("bounded read fits u64"))
            .read_to_end(&mut bytes)
            .map_err(|error| CliError::message(format!("cannot read nofollow file: {error}")))?;
        if bytes.len() > limit {
            return Err(CliError::message("nofollow file exceeds its native bound"));
        }
        Ok(bytes)
    }
    #[cfg(not(unix))]
    {
        let metadata = fs::symlink_metadata(path)
            .map_err(|error| CliError::message(format!("cannot inspect nofollow file: {error}")))?;
        if metadata.file_type().is_symlink() || !metadata.is_file() {
            return Err(CliError::message(
                "nofollow file must be a regular non-symlink file",
            ));
        }
        let bytes = fs::read(path)
            .map_err(|error| CliError::message(format!("cannot read nofollow file: {error}")))?;
        if bytes.len() > limit {
            return Err(CliError::message("nofollow file exceeds its native bound"));
        }
        Ok(bytes)
    }
}

fn required_native_env(name: &str) -> Result<String, CliError> {
    let value = std::env::var(name)
        .map_err(|_| CliError::message(format!("native broker metadata {name} is absent")))?;
    if value.is_empty() || value.chars().any(char::is_control) {
        return Err(CliError::message(format!(
            "native broker metadata {name} is invalid"
        )));
    }
    Ok(value)
}

fn read_json<T: for<'de> Deserialize<'de>>(context: &mut ExecutionContext) -> Result<T, CliError> {
    let mut input = String::new();
    loop {
        let before = input.len();
        context.read_stdin(&mut input).map_err(|error| {
            CliError::message(format!("cannot read native broker input: {error}"))
        })?;
        if input.len() > MAX_INPUT_BYTES {
            return Err(CliError::message("native broker input exceeds its bound"));
        }
        if input.len() == before {
            break;
        }
    }
    serde_json::from_str(&input).map_err(|error| {
        CliError::message(format!("native broker input is not strict JSON: {error}"))
    })
}

fn write_json<T: Serialize>(context: &mut ExecutionContext, value: &T) -> Result<(), CliError> {
    let mut bytes = serde_json::to_vec(value).map_err(|error| {
        CliError::message(format!("cannot encode native broker response: {error}"))
    })?;
    bytes.push(b'\n');
    context
        .write_stdout(&bytes)
        .map_err(|error| CliError::message(format!("cannot write native broker response: {error}")))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn native_agent_types_use_the_same_role_for_plain_and_canonical_carriers() {
        for harness in [Harness::Codex, Harness::Pi, Harness::ClaudeCode] {
            for role in [
                Role::Engineer,
                Role::Coder,
                Role::Worker,
                Role::Conductor,
                Role::Auditor,
                Role::Critic,
                Role::Discovery,
            ] {
                let expected = match harness {
                    Harness::Codex => {
                        if role.write_eligible() {
                            "worker".into()
                        } else {
                            "explorer".into()
                        }
                    }
                    Harness::Pi => format!("pi-subagents:{}", role.as_str()),
                    _ => role.as_str().to_owned(),
                };
                let carrier = role.carrier();
                for spelling in [role.as_str(), carrier.as_str()] {
                    assert_eq!(
                        native_agent_type(harness, parse_role(spelling).expect("recognized role")),
                        expected,
                        "{harness:?}: {spelling}"
                    );
                }
            }
        }
    }

    #[test]
    fn launch_contract_compares_typed_roles_without_rewriting_the_request() {
        let executable = std::env::current_exe().expect("test executable");
        let project_root = executable.parent().expect("fixture root");
        let mut request = NativeLaunchRequest {
            schema: LAUNCH_SCHEMA.into(),
            harness: Harness::Codex,
            parent_role: "shepherd:shepherd".into(),
            parent_session_id: "root".into(),
            root_session_id: "root".into(),
            parent_dispatch_id: None,
            prepare: PreparePendingDispatchRequest {
                schema: "shepherd.pending-dispatch-request/2".into(),
                run: Some("v657".into()),
                role: "engineer".into(),
                work_kind: "planning".into(),
                lane: None,
                parent_dispatch_id: None,
                replaces_agent_id: None,
                baseline: "a".repeat(40),
                read_scope: vec!["docs/task.md".into()],
                write_scope: vec![".shepherd/runs/v657/plan.md".into()],
                result_artifact: ".shepherd/runs/v657/plan.md".into(),
                review_artifact: ".shepherd/runs/v657/reports/engineer.json".into(),
                task_file: "docs/task.md".into(),
                child_session_id: "child".into(),
                lease_ms: 60_000,
                expected_attachment: crate::CarrierAttachmentExpectationRequest {
                    target: Harness::Codex,
                    role: "shepherd:engineer".into(),
                    agent_id: "engineer".into(),
                    attachment_kind: "codex-custom-agent".into(),
                },
            },
            executable: executable.to_str().expect("UTF-8 executable").into(),
            args: vec![],
            env: vec![],
            auth_snapshot: None,
            timeout_ms: 60_000,
        };
        for harness in [Harness::Codex, Harness::Pi, Harness::ClaudeCode] {
            request.harness = harness;
            request.prepare.expected_attachment.target = harness;
            request.prepare.expected_attachment.attachment_kind = match harness {
                Harness::Codex => "codex-custom-agent",
                Harness::Pi => "pi-skill-path",
                _ => "claude-preload",
            }
            .into();
            for (plain, canonical) in [
                ("engineer", "shepherd:engineer"),
                ("shepherd:engineer", "engineer"),
            ] {
                request.prepare.role = plain.into();
                request.prepare.expected_attachment.role = canonical.into();
                assert!(
                    validate_launch_request(&request, project_root).is_ok(),
                    "{harness:?}: {plain}/{canonical}"
                );
                assert_eq!(request.prepare.role, plain);
                assert_eq!(request.prepare.expected_attachment.role, canonical);
            }
        }
        request.prepare.expected_attachment.role = "coder".into();
        assert!(
            validate_launch_request(&request, project_root).is_err(),
            "different role remains denied"
        );
        for invalid in ["shepherd:bogus", "engineer;echo unsafe", "coder/worker", ""] {
            request.prepare.role = invalid.into();
            request.prepare.expected_attachment.role = invalid.into();
            assert!(
                validate_launch_request(&request, project_root).is_err(),
                "unrecognized role remains denied"
            );
        }
    }

    #[test]
    fn launch_contract_rejects_ambient_identity_environment_names() {
        for name in [
            "PATH",
            "PI_SUBAGENT_RUN_ID",
            "CLAUDE_SESSION_ID",
            "SHEPHERD_ROLE",
        ] {
            let entry = LaunchEnvironmentEntry {
                name: name.into(),
                value: "ambient".into(),
            };
            let result = validate_environment_entry(&entry);
            if name == "PATH" {
                assert!(result.is_ok(), "PATH may be explicit, never ambient");
            } else {
                assert!(result.is_err(), "{name} must be rejected");
            }
        }
    }

    #[cfg(unix)]
    #[test]
    fn long_host_temp_paths_fall_back_before_unix_socket_allocation() {
        let long = PathBuf::from("/tmp").join("x".repeat(180));
        assert_eq!(endpoint_root(&long), PathBuf::from("/tmp"));
        assert_eq!(endpoint_root(Path::new("/tmp")), PathBuf::from("/tmp"));
    }

    #[test]
    fn child_event_contract_has_no_status_or_path_fields() {
        let value = serde_json::to_value(NativeChildEventRequest {
            schema: CHILD_EVENT_SCHEMA.into(),
            harness: Harness::Pi,
            event: ChildEvent::Start,
            session_id: "child-session".into(),
            agent_id: "child-agent".into(),
            agent_type: "pi-subagents:worker".into(),
        })
        .expect("child event JSON");
        assert!(value.get("status").is_none());
        assert!(value.get("result_artifact").is_none());
        assert!(value.get("nonce").is_none());
    }
}