shepherd-cli 6.6.1

//! OS-authenticated native launch broker.
//!
//! The broker is the only authority that owns launch secrets.  The dispatch
//! ledger stores only fixed-width hashes; clients receive an opaque launch
//! handle and must return over the private broker channel.  A socket path or a
//! JSON file is never treated as a bearer credential.

use std::{
    fmt, io,
    path::{Path, PathBuf},
};

use sha2::{Digest, Sha256};

use shepherd::{
    Harness,
    dispatch::{
        CarrierAttachmentExpectation, DispatchId, DispatchRecord, PendingLaunchState, Role, RunId,
        SessionId, constant_time_digest_eq,
    },
};

pub const BROKER_PROTOCOL_SCHEMA: &str = "shepherd.native-broker/1";
pub const BROKER_COMPLETION_SCHEMA: &str = "shepherd.native-broker-completion/1";
const BROKER_SECRET_BYTES: usize = 32;

#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct BrokerParent {
    pub(crate) process: ProcessIdentity,
    pub(crate) harness: Harness,
    pub(crate) role: Role,
    pub(crate) session_id: SessionId,
    pub(crate) root_session_id: SessionId,
    pub(crate) dispatch_id: Option<DispatchId>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct BrokerChild {
    /// The registered harness identity, not the hook helper's identity. Hook
    /// helpers are descendants and may submit events, but durable custody is
    /// always bound to the real harness process that the parent registered.
    pub(crate) process: ProcessIdentity,
    pub(crate) parent_process_hash: [u8; 32],
    pub(crate) launch_id: BrokerLaunchId,
    pub(crate) launch_hash: [u8; 32],
    pub(crate) run: RunId,
    pub(crate) nonce_sha256: [u8; 32],
    pub(crate) expected_attachment: CarrierAttachmentExpectation,
    pub(crate) initial_connection: bool,
}
type BrokerResult<T> = Result<T, BrokerError>;

#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum BrokerError {
    #[error("native broker I/O failed: {0}")]
    Io(#[source] io::Error),
    #[error("native broker protocol error: {0}")]
    Protocol(String),
    #[error("native broker peer credentials are unavailable")]
    PeerCredentialsUnavailable,
    #[error("native broker peer is not the registered process")]
    WrongPeer,
    #[error("native broker peer process identity is stale")]
    StalePeer,
    #[error("native broker child is not a descendant of its registered parent")]
    WrongAncestry,
    #[error("native broker launch identity was replayed or consumed")]
    Replay,
    #[error("native broker does not support this platform")]
    UnsupportedPlatform,
}

impl From<io::Error> for BrokerError {
    fn from(error: io::Error) -> Self {
        Self::Io(error)
    }
}

#[derive(Clone, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct ProcessIdentity {
    pid: u32,
    uid: u32,
    start_hash: [u8; 32],
}

impl ProcessIdentity {
    pub fn current() -> BrokerResult<Self> {
        Self::for_pid(std::process::id())
    }

    pub fn for_pid(pid: u32) -> BrokerResult<Self> {
        if pid == 0 {
            return Err(BrokerError::Protocol("process id must be non-zero".into()));
        }
        let (uid, start_material) = process_material(pid)?;
        let start_hash = digest(&start_material);
        if start_hash == [0; 32] {
            return Err(BrokerError::StalePeer);
        }
        Ok(Self {
            pid,
            uid,
            start_hash,
        })
    }

    #[must_use]
    pub const fn pid(self) -> u32 {
        self.pid
    }

    #[must_use]
    pub const fn uid(self) -> u32 {
        self.uid
    }

    #[must_use]
    pub fn digest(self) -> [u8; 32] {
        let mut bytes = [0_u8; 40];
        bytes[..4].copy_from_slice(&self.pid.to_be_bytes());
        bytes[4..8].copy_from_slice(&self.uid.to_be_bytes());
        bytes[8..].copy_from_slice(&self.start_hash);
        digest(&bytes)
    }

    #[must_use]
    pub fn same_process(self, other: Self) -> bool {
        self.pid == other.pid
            && self.uid == other.uid
            && constant_time_digest_eq(&self.start_hash, &other.start_hash)
    }
}

impl fmt::Debug for ProcessIdentity {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ProcessIdentity")
            .field("pid", &self.pid)
            .field("uid", &self.uid)
            .field("start_hash", &"<redacted>")
            .finish()
    }
}

#[derive(Clone, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub(crate) struct BrokerSecret(pub(crate) [u8; BROKER_SECRET_BYTES]);

impl fmt::Debug for BrokerSecret {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("BrokerSecret(<redacted>)")
    }
}

impl Drop for BrokerSecret {
    fn drop(&mut self) {
        self.0.fill(0);
    }
}

#[derive(Clone, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct BrokerLaunchId([u8; BROKER_SECRET_BYTES]);

impl BrokerLaunchId {
    fn from_bytes(bytes: [u8; BROKER_SECRET_BYTES]) -> Self {
        Self(bytes)
    }

    pub(crate) fn fresh() -> BrokerResult<Self> {
        Ok(Self(os_random_32()?))
    }

    #[must_use]
    pub fn hash(self) -> [u8; 32] {
        digest(&self.0)
    }

    /// Encode the non-secret correlation id for a child callback. Authority is
    /// still the authenticated peer plus the broker's in-memory lease.
    #[must_use]
    pub fn opaque_string(self) -> String {
        hex_bytes(&self.0)
    }

    pub fn from_opaque(value: &str) -> BrokerResult<Self> {
        if value.len() != BROKER_SECRET_BYTES * 2 {
            return Err(BrokerError::Protocol(
                "opaque launch id length is invalid".into(),
            ));
        }
        let mut bytes = [0_u8; BROKER_SECRET_BYTES];
        for (index, pair) in value.as_bytes().as_chunks::<2>().0.iter().enumerate() {
            let high = broker_hex_digit(pair[0]).ok_or_else(|| {
                BrokerError::Protocol("opaque launch id is not hexadecimal".into())
            })?;
            let low = broker_hex_digit(pair[1]).ok_or_else(|| {
                BrokerError::Protocol("opaque launch id is not hexadecimal".into())
            })?;
            bytes[index] = high << 4 | low;
        }
        Ok(Self(bytes))
    }
}

impl fmt::Debug for BrokerLaunchId {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("BrokerLaunchId(<opaque>)")
    }
}

impl fmt::Display for BrokerLaunchId {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str("<opaque-launch-id>")
    }
}

pub(crate) fn os_random_32() -> BrokerResult<[u8; BROKER_SECRET_BYTES]> {
    let mut bytes = [0_u8; BROKER_SECRET_BYTES];
    getrandom::fill(&mut bytes)
        .map_err(|error| BrokerError::Io(io::Error::other(error.to_string())))?;
    Ok(bytes)
}

/// Return only the digest of fresh random material. The random bytes exist
/// solely in broker memory and never enter a durable request or artifact.
pub(crate) fn random_digest_32() -> BrokerResult<[u8; 32]> {
    let secret = BrokerSecret(os_random_32()?);
    Ok(digest(&secret.0))
}

pub(crate) fn digest(bytes: &[u8]) -> [u8; 32] {
    Sha256::digest(bytes).into()
}

pub(crate) fn hex_bytes(bytes: &[u8]) -> String {
    let mut output = String::with_capacity(bytes.len() * 2);
    for byte in bytes {
        output.push_str(&format!("{byte:02x}"));
    }
    output
}

fn broker_hex_digit(value: u8) -> Option<u8> {
    match value {
        b'0'..=b'9' => Some(value - b'0'),
        b'a'..=b'f' => Some(value - b'a' + 10),
        b'A'..=b'F' => Some(value - b'A' + 10),
        _ => None,
    }
}

#[cfg(unix)]
#[allow(unsafe_code)]
fn process_material(pid: u32) -> BrokerResult<(u32, Vec<u8>)> {
    let uid = unsafe { libc::geteuid() };
    #[cfg(target_os = "linux")]
    {
        let stat = std::fs::read_to_string(format!("/proc/{pid}/stat"))?;
        let end = stat.rfind(')').ok_or(BrokerError::StalePeer)?;
        let fields: Vec<&str> = stat[end + 1..].split_whitespace().collect();
        let start = fields.get(19).ok_or(BrokerError::StalePeer)?;
        Ok((uid, start.as_bytes().to_vec()))
    }
    #[cfg(target_os = "macos")]
    {
        let output = std::process::Command::new("/bin/ps")
            .args(["-o", "lstart=", "-p", &pid.to_string()])
            .output()?;
        if !output.status.success() || output.stdout.is_empty() {
            return Err(BrokerError::StalePeer);
        }
        Ok((uid, output.stdout))
    }
    #[cfg(not(any(target_os = "linux", target_os = "macos")))]
    {
        let result = unsafe { libc::kill(pid.cast_signed(), 0) };
        if result != 0 {
            return Err(BrokerError::StalePeer);
        }
        Ok((uid, pid.to_be_bytes().to_vec()))
    }
}

#[cfg(windows)]
#[allow(unsafe_code)]
fn process_material(pid: u32) -> BrokerResult<(u32, Vec<u8>)> {
    use std::ptr::null_mut;
    use windows_sys::Win32::{
        Foundation::{CloseHandle, FILETIME},
        Security::{GetLengthSid, GetTokenInformation, TOKEN_QUERY, TOKEN_USER, TokenUser},
        System::Threading::{
            GetProcessTimes, OpenProcess, OpenProcessToken, PROCESS_QUERY_LIMITED_INFORMATION,
        },
    };

    let process = unsafe { OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
    if process.is_null() {
        return Err(BrokerError::StalePeer);
    }
    let mut creation = FILETIME::default();
    let mut exit = FILETIME::default();
    let mut kernel = FILETIME::default();
    let mut user = FILETIME::default();
    let times_ok =
        unsafe { GetProcessTimes(process, &mut creation, &mut exit, &mut kernel, &mut user) } != 0;
    if !times_ok {
        unsafe { CloseHandle(process) };
        return Err(BrokerError::StalePeer);
    }

    let mut token = null_mut();
    let token_ok = unsafe { OpenProcessToken(process, TOKEN_QUERY, &mut token) } != 0;
    if !token_ok {
        unsafe { CloseHandle(process) };
        return Err(BrokerError::PeerCredentialsUnavailable);
    }
    let mut required = 0_u32;
    unsafe {
        GetTokenInformation(token, TokenUser, null_mut(), 0, &mut required);
    }
    if required == 0 {
        unsafe {
            CloseHandle(token);
            CloseHandle(process);
        }
        return Err(BrokerError::PeerCredentialsUnavailable);
    }
    let mut buffer = vec![0_u8; usize::try_from(required).expect("Windows token size fits usize")];
    let info_ok = unsafe {
        GetTokenInformation(
            token,
            TokenUser,
            buffer.as_mut_ptr().cast(),
            required,
            &mut required,
        )
    } != 0;
    if !info_ok {
        unsafe {
            CloseHandle(token);
            CloseHandle(process);
        }
        return Err(BrokerError::PeerCredentialsUnavailable);
    }
    let sid = unsafe { (*(buffer.as_ptr().cast::<TOKEN_USER>())).User.Sid };
    if sid.is_null() {
        unsafe {
            CloseHandle(token);
            CloseHandle(process);
        }
        return Err(BrokerError::PeerCredentialsUnavailable);
    }
    let sid_length = unsafe { GetLengthSid(sid) };
    if sid_length == 0 {
        unsafe {
            CloseHandle(token);
            CloseHandle(process);
        }
        return Err(BrokerError::PeerCredentialsUnavailable);
    }
    let sid_bytes = unsafe {
        std::slice::from_raw_parts(
            sid.cast::<u8>(),
            usize::try_from(sid_length).expect("Windows SID size fits usize"),
        )
    };
    let uid_digest = digest(sid_bytes);
    let uid = u32::from_be_bytes(uid_digest[..4].try_into().expect("digest prefix"));
    let mut start_material = Vec::with_capacity(8);
    start_material.extend_from_slice(&creation.dwHighDateTime.to_be_bytes());
    start_material.extend_from_slice(&creation.dwLowDateTime.to_be_bytes());
    unsafe {
        CloseHandle(token);
        CloseHandle(process);
    }
    Ok((uid, start_material))
}

#[cfg(not(any(unix, windows)))]
fn process_material(_pid: u32) -> BrokerResult<(u32, Vec<u8>)> {
    Err(BrokerError::UnsupportedPlatform)
}

#[cfg(unix)]
pub(crate) fn peer_identity(
    stream: &std::os::unix::net::UnixStream,
) -> BrokerResult<ProcessIdentity> {
    use std::os::fd::AsRawFd;

    let fd = stream.as_raw_fd();
    let (pid, uid) = peer_credentials(fd)?;
    let identity = ProcessIdentity::for_pid(pid)?;
    if identity.uid != uid {
        return Err(BrokerError::WrongPeer);
    }
    Ok(identity)
}

#[cfg(unix)]
#[allow(unsafe_code)]
fn peer_credentials(fd: std::os::unix::io::RawFd) -> BrokerResult<(u32, u32)> {
    #[cfg(target_os = "linux")]
    {
        let mut credentials = libc::ucred {
            pid: 0,
            uid: 0,
            gid: 0,
        };
        let mut length = libc::socklen_t::try_from(std::mem::size_of::<libc::ucred>())
            .expect("ucred socket length fits socklen_t");
        let result = unsafe {
            libc::getsockopt(
                fd,
                libc::SOL_SOCKET,
                libc::SO_PEERCRED,
                (&mut credentials as *mut libc::ucred).cast(),
                &mut length,
            )
        };
        if result != 0 || credentials.pid <= 0 {
            return Err(BrokerError::PeerCredentialsUnavailable);
        }
        Ok((
            u32::try_from(credentials.pid).map_err(|_| BrokerError::StalePeer)?,
            credentials.uid,
        ))
    }
    #[cfg(target_os = "macos")]
    {
        let mut uid = 0_u32;
        let mut gid = 0_u32;
        let result = unsafe { libc::getpeereid(fd, &mut uid, &mut gid) };
        if result != 0 {
            return Err(BrokerError::PeerCredentialsUnavailable);
        }
        let mut pid = 0_i32;
        let mut length = libc::socklen_t::try_from(std::mem::size_of::<libc::pid_t>())
            .expect("pid socket length fits socklen_t");
        let result = unsafe {
            libc::getsockopt(
                fd,
                libc::SOL_LOCAL,
                libc::LOCAL_PEEREPID,
                (&mut pid as *mut libc::pid_t).cast(),
                &mut length,
            )
        };
        if result != 0 || pid <= 0 {
            return Err(BrokerError::PeerCredentialsUnavailable);
        }
        Ok((u32::try_from(pid).map_err(|_| BrokerError::StalePeer)?, uid))
    }
    #[cfg(not(any(target_os = "linux", target_os = "macos")))]
    {
        let _ = fd;
        Err(BrokerError::PeerCredentialsUnavailable)
    }
}

#[cfg(unix)]
pub(crate) fn is_descendant(child: ProcessIdentity, parent: ProcessIdentity) -> BrokerResult<bool> {
    if child.uid != parent.uid || child.pid == parent.pid {
        return Ok(false);
    }
    let mut current = child.pid;
    for _ in 0..64 {
        let Some((_, ppid)) = process_parent(current)? else {
            return Ok(false);
        };
        if ppid == parent.pid {
            return Ok(true);
        }
        if ppid == 0 || ppid == current {
            return Ok(false);
        }
        current = ppid;
    }
    Ok(false)
}

#[cfg(unix)]
fn process_parent(pid: u32) -> BrokerResult<Option<(u32, u32)>> {
    #[cfg(target_os = "linux")]
    {
        let stat = match std::fs::read_to_string(format!("/proc/{pid}/stat")) {
            Ok(stat) => stat,
            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
            Err(error) => return Err(error.into()),
        };
        let end = stat.rfind(')').ok_or(BrokerError::StalePeer)?;
        let fields: Vec<&str> = stat[end + 1..].split_whitespace().collect();
        let ppid = fields
            .get(1)
            .ok_or(BrokerError::StalePeer)?
            .parse::<u32>()
            .map_err(|_| BrokerError::StalePeer)?;
        Ok(Some((pid, ppid)))
    }
    #[cfg(target_os = "macos")]
    {
        let output = std::process::Command::new("/bin/ps")
            .args(["-o", "ppid=", "-p", &pid.to_string()])
            .output()?;
        if !output.status.success() {
            return Ok(None);
        }
        let ppid = String::from_utf8_lossy(&output.stdout)
            .trim()
            .parse::<u32>()
            .map_err(|_| BrokerError::StalePeer)?;
        Ok(Some((pid, ppid)))
    }
    #[cfg(not(any(target_os = "linux", target_os = "macos")))]
    {
        let _ = pid;
        Ok(None)
    }
}

#[cfg(windows)]
#[allow(unsafe_code)]
pub(crate) fn is_descendant(child: ProcessIdentity, parent: ProcessIdentity) -> BrokerResult<bool> {
    use std::collections::HashMap;
    use windows_sys::Win32::{
        Foundation::{CloseHandle, INVALID_HANDLE_VALUE},
        System::Diagnostics::ToolHelp::{
            CreateToolhelp32Snapshot, PROCESSENTRY32W, Process32FirstW, Process32NextW,
            TH32CS_SNAPPROCESS,
        },
    };

    if child.uid != parent.uid || child.pid == parent.pid {
        return Ok(false);
    }
    let snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0) };
    if snapshot == INVALID_HANDLE_VALUE {
        return Err(BrokerError::Io(io::Error::last_os_error()));
    }
    let mut parents = HashMap::new();
    let mut entry = PROCESSENTRY32W {
        dwSize: u32::try_from(std::mem::size_of::<PROCESSENTRY32W>())
            .expect("Windows process entry size fits u32"),
        ..Default::default()
    };
    let mut enumerated = unsafe { Process32FirstW(snapshot, &mut entry) } != 0;
    while enumerated {
        parents.insert(entry.th32ProcessID, entry.th32ParentProcessID);
        enumerated = unsafe { Process32NextW(snapshot, &mut entry) } != 0;
    }
    unsafe { CloseHandle(snapshot) };
    if parents.is_empty() {
        return Err(BrokerError::StalePeer);
    }
    let mut current = child.pid;
    for _ in 0..64 {
        let Some(&ppid) = parents.get(&current) else {
            return Ok(false);
        };
        if ppid == parent.pid {
            return Ok(true);
        }
        if ppid == 0 || ppid == current {
            return Ok(false);
        }
        current = ppid;
    }
    Ok(false)
}

use crate::DispatchService;
#[cfg(any(unix, windows))]
use crate::{ClaimPendingDispatchRequest, PreparePendingDispatchRequest};
#[cfg(any(unix, windows))]
use shepherd::dispatch::LoadedCarrierAttestationV1;

#[cfg(any(unix, windows))]
use std::collections::HashMap;

#[cfg(unix)]
use std::{
    fs::{self, Permissions},
    os::unix::fs::{FileTypeExt, MetadataExt, PermissionsExt},
    os::unix::net::{UnixListener, UnixStream},
    sync::{
        Arc, Mutex,
        atomic::{AtomicBool, Ordering},
    },
    thread::{self, JoinHandle},
};

#[cfg(windows)]
use std::{
    ffi::OsStr,
    os::windows::ffi::OsStrExt,
    sync::{
        Arc, Mutex,
        atomic::{AtomicBool, Ordering},
    },
    thread::{self, JoinHandle},
    time::{Duration, Instant},
};

#[derive(Clone, Eq, PartialEq)]
pub struct LaunchHandle {
    endpoint: PathBuf,
    launch_id: BrokerLaunchId,
    run: RunId,
    nonce_sha256: [u8; 32],
}

impl fmt::Debug for LaunchHandle {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("LaunchHandle")
            .field("endpoint", &self.endpoint)
            .field("launch_id", &"<opaque>")
            .field("run", &self.run)
            .finish()
    }
}

impl LaunchHandle {
    #[must_use]
    pub fn endpoint(&self) -> &Path {
        &self.endpoint
    }

    #[must_use]
    pub fn launch_id(&self) -> BrokerLaunchId {
        self.launch_id
    }

    /// The broker challenge digest is a durable hash fact, not secret launch
    /// material. It is transported over the authenticated broker channel so a
    /// provider never needs a pending JSON path to build its attestation.
    #[must_use]
    pub fn nonce_sha256(&self) -> [u8; 32] {
        self.nonce_sha256
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ClaimedLaunch {
    pub launch_id: BrokerLaunchId,
    pub state: shepherd::dispatch::PendingLaunchState,
}

#[cfg(any(unix, windows))]
struct LaunchRecord {
    secret: BrokerSecret,
    parent: BrokerParent,
    run: RunId,
    nonce_sha256: [u8; 32],
    expected_attachment: CarrierAttachmentExpectation,
    child: Option<ProcessIdentity>,
    child_connected: bool,
    activated: bool,
    completed: bool,
    active_parent: Option<BrokerParent>,
}

#[cfg(any(unix, windows))]
struct BrokerState {
    owner: ProcessIdentity,
    launches: HashMap<BrokerLaunchId, LaunchRecord>,
}

/// A private, per-project native broker.  `start` creates a mode-0700 parent
/// directory and a mode-0600 Unix socket.  The endpoint name is discovery
/// data, not authority: every request still requires kernel peer credentials,
/// a registered process start identity, and a one-shot launch entry.
pub struct NativeBroker {
    endpoint: PathBuf,
    #[cfg(any(unix, windows))]
    stop: Arc<AtomicBool>,
    #[cfg(any(unix, windows))]
    thread: Option<JoinHandle<()>>,
}

impl NativeBroker {
    #[cfg(unix)]
    pub fn start(service: DispatchService, endpoint: impl AsRef<Path>) -> BrokerResult<Self> {
        let endpoint = endpoint.as_ref().to_path_buf();
        let owner = ProcessIdentity::current()?;
        let parent = endpoint
            .parent()
            .ok_or_else(|| BrokerError::Protocol("broker endpoint has no private parent".into()))?;
        if parent.exists() {
            let metadata = fs::symlink_metadata(parent)?;
            if metadata.file_type().is_symlink() || !metadata.is_dir() {
                return Err(BrokerError::WrongPeer);
            }
            let owner_uid = owner.uid();
            if metadata.uid() != owner_uid {
                return Err(BrokerError::WrongPeer);
            }
            let mode = metadata.permissions().mode() & 0o777;
            if mode & 0o077 != 0 {
                return Err(BrokerError::WrongPeer);
            }
        } else {
            fs::create_dir_all(parent)?;
            fs::set_permissions(parent, Permissions::from_mode(0o700))?;
        }
        let parent_metadata = fs::symlink_metadata(parent)?;
        if parent_metadata.file_type().is_symlink()
            || !parent_metadata.is_dir()
            || parent_metadata.uid() != owner.uid()
            || parent_metadata.permissions().mode() & 0o077 != 0
        {
            return Err(BrokerError::WrongPeer);
        }
        if fs::symlink_metadata(&endpoint).is_ok() {
            return Err(BrokerError::Protocol(
                "broker endpoint already exists; refusing replacement".into(),
            ));
        }
        match service.store().reconcile_all_unspawned() {
            Ok(_) => {}
            Err(crate::DispatchStoreError::Io { source, .. })
                if source.kind() == io::ErrorKind::NotFound => {}
            Err(error) => return Err(BrokerError::Protocol(error.to_string())),
        }
        let listener = UnixListener::bind(&endpoint)?;
        fs::set_permissions(&endpoint, Permissions::from_mode(0o600))?;
        let endpoint_metadata = fs::symlink_metadata(&endpoint)?;
        if endpoint_metadata.uid() != owner.uid()
            || endpoint_metadata.file_type().is_symlink()
            || !endpoint_metadata.file_type().is_socket()
        {
            let _ = fs::remove_file(&endpoint);
            return Err(BrokerError::WrongPeer);
        }
        let stop = Arc::new(AtomicBool::new(false));
        let thread_stop = Arc::clone(&stop);
        let state = Arc::new(Mutex::new(BrokerState {
            owner,
            launches: HashMap::new(),
        }));
        let thread_state = Arc::clone(&state);
        let thread_endpoint = endpoint.clone();
        let thread = thread::Builder::new()
            .name("shepherd-native-broker".into())
            .spawn(move || {
                while !thread_stop.load(Ordering::Acquire) {
                    match listener.accept() {
                        Ok((stream, _)) => {
                            let service = service.clone();
                            let state = Arc::clone(&thread_state);
                            let _ = thread::Builder::new()
                                .name("shepherd-native-broker-peer".into())
                                .spawn(move || handle_connection(stream, service, state));
                        }
                        Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
                        Err(_) if thread_stop.load(Ordering::Acquire) => break,
                        Err(_) => break,
                    }
                }
                let _ = fs::remove_file(thread_endpoint);
            })
            .map_err(BrokerError::Io)?;
        Ok(Self {
            endpoint,
            stop,
            thread: Some(thread),
        })
    }

    #[cfg(windows)]
    pub fn start(service: DispatchService, endpoint: impl AsRef<Path>) -> BrokerResult<Self> {
        let endpoint = windows_pipe_name(endpoint.as_ref());
        let owner = ProcessIdentity::current()?;
        match service.store().reconcile_all_unspawned() {
            Ok(_) => {}
            Err(crate::DispatchStoreError::Io { source, .. })
                if source.kind() == io::ErrorKind::NotFound => {}
            Err(error) => return Err(BrokerError::Protocol(error.to_string())),
        }
        let first = create_acl_named_pipe(&endpoint, true)?;
        let stop = Arc::new(AtomicBool::new(false));
        let thread_stop = Arc::clone(&stop);
        let state = Arc::new(Mutex::new(BrokerState {
            owner,
            launches: HashMap::new(),
        }));
        let thread_state = Arc::clone(&state);
        let thread_endpoint = endpoint.clone();
        let thread = thread::Builder::new()
            .name("shepherd-native-broker-pipe".into())
            .spawn(move || {
                windows_pipe_loop(thread_endpoint, first, service, thread_state, thread_stop);
            })
            .map_err(BrokerError::Io)?;
        Ok(Self {
            endpoint: PathBuf::from(endpoint),
            stop,
            thread: Some(thread),
        })
    }

    #[cfg(not(any(unix, windows)))]
    pub fn start(_service: DispatchService, endpoint: impl AsRef<Path>) -> BrokerResult<Self> {
        let _ = endpoint;
        Err(BrokerError::UnsupportedPlatform)
    }

    #[must_use]
    pub fn endpoint(&self) -> &Path {
        &self.endpoint
    }

    #[cfg(any(unix, windows))]
    pub fn connect(&self) -> BrokerResult<BrokerClient> {
        BrokerClient::connect_endpoint(&self.endpoint)
    }
}

#[cfg(windows)]
fn windows_pipe_name(endpoint: &Path) -> String {
    let name = endpoint
        .file_name()
        .and_then(|value| value.to_str())
        .filter(|value| !value.is_empty())
        .unwrap_or("broker");
    format!(r"\\.\pipe\shepherd-{name}")
}

#[cfg(windows)]
struct WindowsPipeStream {
    handle: windows_sys::Win32::Foundation::HANDLE,
    server: bool,
    client_pid: u32,
}

#[cfg(windows)]
// Windows kernel handles are process-local capabilities. Moving this wrapper
// between the accept and connection threads does not duplicate or serialize
// the handle; ownership remains with exactly one wrapper.
#[allow(unsafe_code)]
unsafe impl Send for WindowsPipeStream {}

#[cfg(windows)]
impl Drop for WindowsPipeStream {
    #[allow(unsafe_code)]
    fn drop(&mut self) {
        use windows_sys::Win32::{Foundation::CloseHandle, System::Pipes::DisconnectNamedPipe};
        if self.server {
            unsafe { DisconnectNamedPipe(self.handle) };
        }
        unsafe { CloseHandle(self.handle) };
    }
}

#[cfg(windows)]
#[allow(unsafe_code)]
fn create_acl_named_pipe(name: &str, first: bool) -> BrokerResult<WindowsPipeStream> {
    use std::ptr::null_mut;
    use windows_sys::{
        Win32::{
            Foundation::INVALID_HANDLE_VALUE,
            Security::{
                Authorization::{
                    ConvertStringSecurityDescriptorToSecurityDescriptorW, SDDL_REVISION_1,
                },
                SECURITY_ATTRIBUTES,
            },
            Storage::FileSystem::{FILE_FLAG_FIRST_PIPE_INSTANCE, PIPE_ACCESS_DUPLEX},
            System::Pipes::{CreateNamedPipeW, PIPE_READMODE_BYTE, PIPE_TYPE_BYTE, PIPE_WAIT},
        },
        core::PCWSTR,
    };
    let wide_name: Vec<u16> = OsStr::new(name).encode_wide().chain([0]).collect();
    let sddl: Vec<u16> = OsStr::new("D:(A;;GA;;;OW)")
        .encode_wide()
        .chain([0])
        .collect();
    let mut descriptor = null_mut();
    let mut descriptor_size = 0_u32;
    if unsafe {
        ConvertStringSecurityDescriptorToSecurityDescriptorW(
            sddl.as_ptr(),
            SDDL_REVISION_1,
            &mut descriptor,
            &mut descriptor_size,
        )
    } == 0
    {
        return Err(BrokerError::Io(io::Error::last_os_error()));
    }
    let attributes = SECURITY_ATTRIBUTES {
        nLength: u32::try_from(std::mem::size_of::<SECURITY_ATTRIBUTES>())
            .expect("security attributes size fits u32"),
        lpSecurityDescriptor: descriptor,
        bInheritHandle: 0,
    };
    let first_flag = if first {
        FILE_FLAG_FIRST_PIPE_INSTANCE
    } else {
        0
    };
    let handle = unsafe {
        CreateNamedPipeW(
            wide_name.as_ptr() as PCWSTR,
            PIPE_ACCESS_DUPLEX | first_flag,
            PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,
            255,
            1_048_576,
            1_048_576,
            5_000,
            &attributes,
        )
    };
    unsafe { windows_sys::Win32::Foundation::LocalFree(descriptor) };
    if handle == INVALID_HANDLE_VALUE {
        return Err(BrokerError::Io(io::Error::last_os_error()));
    }
    Ok(WindowsPipeStream {
        handle,
        server: true,
        client_pid: 0,
    })
}

#[cfg(windows)]
#[allow(unsafe_code)]
fn connect_windows_pipe(endpoint: &Path) -> BrokerResult<WindowsPipeStream> {
    use windows_sys::Win32::{
        Foundation::{
            ERROR_FILE_NOT_FOUND, ERROR_PIPE_BUSY, GENERIC_READ, GENERIC_WRITE, GetLastError,
            INVALID_HANDLE_VALUE,
        },
        Storage::FileSystem::{CreateFileW, FILE_ATTRIBUTE_NORMAL, FILE_SHARE_NONE, OPEN_EXISTING},
    };
    let wide_name: Vec<u16> = OsStr::new(endpoint.to_string_lossy().as_ref())
        .encode_wide()
        .chain([0])
        .collect();
    let deadline = Instant::now() + Duration::from_secs(5);
    loop {
        let handle = unsafe {
            CreateFileW(
                wide_name.as_ptr(),
                GENERIC_READ | GENERIC_WRITE,
                FILE_SHARE_NONE,
                std::ptr::null(),
                OPEN_EXISTING,
                FILE_ATTRIBUTE_NORMAL,
                std::ptr::null_mut(),
            )
        };
        if handle != INVALID_HANDLE_VALUE {
            return Ok(WindowsPipeStream {
                handle,
                server: false,
                client_pid: 0,
            });
        }
        let error = unsafe { GetLastError() };
        if error != ERROR_PIPE_BUSY && error != ERROR_FILE_NOT_FOUND {
            return Err(BrokerError::Io(io::Error::last_os_error()));
        }
        if Instant::now() >= deadline {
            return Err(BrokerError::Io(io::Error::new(
                io::ErrorKind::TimedOut,
                "timed out connecting to native broker named pipe",
            )));
        }
        thread::sleep(Duration::from_millis(10));
    }
}

#[cfg(windows)]
#[allow(unsafe_code)]
fn windows_pipe_loop(
    name: String,
    first: WindowsPipeStream,
    service: DispatchService,
    state: Arc<Mutex<BrokerState>>,
    stop: Arc<AtomicBool>,
) {
    use std::ptr::null_mut;
    use windows_sys::Win32::{
        Foundation::{ERROR_PIPE_CONNECTED, GetLastError},
        System::Pipes::{ConnectNamedPipe, GetNamedPipeClientProcessId},
    };
    let mut next = Some(first);
    while !stop.load(Ordering::Acquire) {
        let mut stream = match next.take() {
            Some(stream) => stream,
            None => match create_acl_named_pipe(&name, false) {
                Ok(stream) => stream,
                Err(_) => break,
            },
        };
        let connected = unsafe { ConnectNamedPipe(stream.handle, null_mut()) };
        if connected == 0 && unsafe { GetLastError() } != ERROR_PIPE_CONNECTED {
            continue;
        }
        let mut client_pid = 0_u32;
        if unsafe { GetNamedPipeClientProcessId(stream.handle, &mut client_pid) } == 0
            || client_pid == 0
        {
            continue;
        }
        stream.client_pid = client_pid;
        let service = service.clone();
        let state = Arc::clone(&state);
        let _ = thread::Builder::new()
            .name("shepherd-native-broker-pipe-peer".into())
            .spawn(move || handle_connection(stream, service, state));
    }
}

#[cfg(unix)]
impl Drop for NativeBroker {
    fn drop(&mut self) {
        self.stop.store(true, Ordering::Release);
        let _ = UnixStream::connect(&self.endpoint);
        if let Some(thread) = self.thread.take() {
            let _ = thread.join();
        }
        let _ = fs::remove_file(&self.endpoint);
    }
}

#[cfg(windows)]
impl Drop for NativeBroker {
    fn drop(&mut self) {
        self.stop.store(true, Ordering::Release);
        let _ = connect_windows_pipe(&self.endpoint);
        if let Some(thread) = self.thread.take() {
            let _ = thread.join();
        }
    }
}

#[cfg(not(any(unix, windows)))]
impl Drop for NativeBroker {
    fn drop(&mut self) {}
}

#[cfg(any(unix, windows))]
pub struct BrokerClient {
    endpoint: PathBuf,
    #[cfg(unix)]
    stream: UnixStream,
    #[cfg(windows)]
    stream: WindowsPipeStream,
}

#[cfg(any(unix, windows))]
impl BrokerClient {
    pub fn connect_endpoint(endpoint: impl AsRef<Path>) -> BrokerResult<Self> {
        let endpoint = endpoint.as_ref().to_path_buf();
        #[cfg(unix)]
        let stream = UnixStream::connect(&endpoint)?;
        #[cfg(windows)]
        let stream = connect_windows_pipe(&endpoint)?;
        Ok(Self { stream, endpoint })
    }

    pub fn connect_child_by_id(
        endpoint: impl AsRef<Path>,
        launch_id: BrokerLaunchId,
    ) -> BrokerResult<BrokerChildClient> {
        let endpoint = endpoint.as_ref().to_path_buf();
        connect_child_stream(
            endpoint.clone(),
            LaunchHandle {
                endpoint,
                launch_id,
                run: RunId::new("broker-pending").expect("internal callback run placeholder"),
                nonce_sha256: [0; 32],
            },
            false,
        )
    }

    /// Connect a later hook helper for a terminal event. The initial child
    /// connection is one-shot; terminal helpers use a distinct event channel
    /// that still authenticates the registered harness ancestry and launch
    /// identity in the broker.
    pub fn connect_child_event_by_id(
        endpoint: impl AsRef<Path>,
        launch_id: BrokerLaunchId,
    ) -> BrokerResult<BrokerChildClient> {
        let endpoint = endpoint.as_ref().to_path_buf();
        connect_child_stream(
            endpoint.clone(),
            LaunchHandle {
                endpoint,
                launch_id,
                run: RunId::new("broker-pending").expect("internal callback run placeholder"),
                nonce_sha256: [0; 32],
            },
            true,
        )
    }

    pub fn register_parent(
        &mut self,
        harness: Harness,
        role: Role,
        session_id: SessionId,
        root_session_id: SessionId,
        dispatch_id: Option<DispatchId>,
    ) -> BrokerResult<()> {
        match self.rpc(BrokerRequest::ParentHello {
            schema: BROKER_PROTOCOL_SCHEMA.into(),
            harness,
            role,
            session_id: session_id.to_string(),
            root_session_id: root_session_id.to_string(),
            dispatch_id: dispatch_id.map(|value| value.to_string()),
        })? {
            BrokerResponse::Ok => Ok(()),
            response => response.into_error(),
        }
    }

    pub fn prepare(
        &mut self,
        request: PreparePendingDispatchRequest,
    ) -> BrokerResult<LaunchHandle> {
        match self.rpc(BrokerRequest::Prepare { request })? {
            BrokerResponse::Prepared {
                launch_id,
                run,
                nonce_sha256,
            } => Ok(LaunchHandle {
                endpoint: self.endpoint.clone(),
                launch_id: BrokerLaunchId::from_bytes(launch_id),
                run: RunId::new(run).map_err(|error| BrokerError::Protocol(error.to_string()))?,
                nonce_sha256,
            }),
            response => response.into_error(),
        }
    }

    pub fn register_child(&mut self, launch: &LaunchHandle, pid: u32) -> BrokerResult<()> {
        match self.rpc(BrokerRequest::RegisterChild {
            launch_id: launch.launch_id.0,
            pid,
        })? {
            BrokerResponse::Ok => Ok(()),
            response => response.into_error(),
        }
    }

    /// Terminalize a launch that never reached activation. The broker, not a
    /// provider hook, supplies the durable launch hash and takes the run lock.
    pub fn cleanup(
        &mut self,
        launch: &LaunchHandle,
        state: PendingLaunchState,
    ) -> BrokerResult<shepherd::dispatch::LaunchCleanupResponse> {
        if !state.is_terminal() {
            return Err(BrokerError::Protocol(
                "broker cleanup requires a terminal launch state".into(),
            ));
        }
        match self.rpc(BrokerRequest::Cleanup {
            launch_id: launch.launch_id.0,
            state,
        })? {
            BrokerResponse::Cleaned { response } => Ok(response),
            response => response.into_error(),
        }
    }

    pub fn connect_child(&self, launch: &LaunchHandle) -> BrokerResult<BrokerChildClient> {
        connect_child_stream(self.endpoint.clone(), launch.clone(), false)
    }

    fn rpc(&mut self, request: BrokerRequest) -> BrokerResult<BrokerResponse> {
        write_frame(&mut self.stream, &request)?;
        read_frame(&mut self.stream)
    }
}

#[cfg(any(unix, windows))]
fn connect_child_stream(
    endpoint: PathBuf,
    launch: LaunchHandle,
    event_connection: bool,
) -> BrokerResult<BrokerChildClient> {
    #[cfg(unix)]
    let stream = UnixStream::connect(&endpoint)?;
    #[cfg(windows)]
    let stream = connect_windows_pipe(&endpoint)?;
    let mut client = BrokerChildClient {
        run: launch.run.clone(),
        launch,
        stream,
        launch_hash: [0; 32],
        nonce_sha256: [0; 32],
        expected_attachment: None,
        event_connection,
    };
    let request = if event_connection {
        BrokerRequest::ChildEventHello {
            launch_id: client.launch.launch_id.0,
        }
    } else {
        BrokerRequest::ChildHello {
            launch_id: client.launch.launch_id.0,
        }
    };
    match client.rpc(request)? {
        BrokerResponse::ChildAccepted {
            run,
            launch_hash,
            nonce_sha256,
            expected_attachment,
        } => {
            client.run =
                RunId::new(run).map_err(|error| BrokerError::Protocol(error.to_string()))?;
            client.launch_hash = launch_hash;
            client.nonce_sha256 = nonce_sha256;
            client.expected_attachment = Some(expected_attachment);
            Ok(client)
        }
        response => response.into_error(),
    }
}

#[cfg(any(unix, windows))]
pub struct BrokerChildClient {
    launch: LaunchHandle,
    #[cfg(unix)]
    stream: UnixStream,
    #[cfg(windows)]
    stream: WindowsPipeStream,
    run: RunId,
    launch_hash: [u8; 32],
    nonce_sha256: [u8; 32],
    expected_attachment: Option<CarrierAttachmentExpectation>,
    event_connection: bool,
}

#[cfg(any(unix, windows))]
impl BrokerChildClient {
    #[must_use]
    pub fn nonce_sha256(&self) -> [u8; 32] {
        self.nonce_sha256
    }

    pub fn expected_attachment(&self) -> BrokerResult<&CarrierAttachmentExpectation> {
        self.expected_attachment.as_ref().ok_or_else(|| {
            BrokerError::Protocol("broker did not provide an attachment expectation".into())
        })
    }

    pub fn loaded_carrier_attestation(
        &self,
    ) -> BrokerResult<shepherd::dispatch::LoadedCarrierAttestationV1> {
        let expected = self.expected_attachment()?.clone();
        Ok(shepherd::dispatch::LoadedCarrierAttestationV1 {
            schema: shepherd::dispatch::LOADED_CARRIER_SCHEMA.into(),
            nonce_sha256: self.nonce_sha256,
            target: expected.target,
            role: expected.role,
            agent_id: expected.agent_id,
            installed_carrier_path: expected.installed_carrier_path,
            candidate_sha256: expected.candidate_sha256,
            carrier_sha256: expected.carrier_sha256,
            compiler_tree_sha256: expected.compiler_tree_sha256,
            startup_skill: expected.startup_skill,
            skill_bundle_sha256: expected.skill_bundle_sha256,
            attachment_kind: expected.attachment_kind,
        })
    }

    pub fn claim(
        &mut self,
        agent_id: String,
        session_id: String,
        agent_type: String,
        attestation: LoadedCarrierAttestationV1,
    ) -> BrokerResult<ClaimedLaunch> {
        match self.rpc(BrokerRequest::Claim {
            launch_id: self.launch.launch_id.0,
            agent_id,
            session_id,
            agent_type,
            attestation,
        })? {
            BrokerResponse::Claimed => Ok(ClaimedLaunch {
                launch_id: self.launch.launch_id,
                state: shepherd::dispatch::PendingLaunchState::ClaimedUnspawned,
            }),
            response => response.into_error(),
        }
    }

    pub fn activate(
        &mut self,
        agent_id: String,
        session_id: String,
        agent_type: String,
        attestation: LoadedCarrierAttestationV1,
    ) -> BrokerResult<shepherd::dispatch::DispatchRecord> {
        match self.rpc(BrokerRequest::Activate {
            launch_id: self.launch.launch_id.0,
            agent_id,
            session_id,
            agent_type,
            attestation,
        })? {
            BrokerResponse::Activated { record } => Ok(record),
            response => response.into_error(),
        }
    }

    pub fn complete(
        &mut self,
        agent_id: String,
        session_id: String,
        agent_type: String,
    ) -> BrokerResult<DispatchRecord> {
        if !self.event_connection {
            return Err(BrokerError::Protocol(
                "broker completion requires an event connection".into(),
            ));
        }
        match self.rpc(BrokerRequest::Complete {
            launch_id: self.launch.launch_id.0,
            agent_id,
            session_id,
            agent_type,
            nonce_sha256: self.nonce_sha256,
        })? {
            BrokerResponse::Completed {
                schema,
                launch_id_hash,
                nonce_sha256,
                record,
            } => {
                if schema != BROKER_COMPLETION_SCHEMA
                    || !constant_time_digest_eq(&launch_id_hash, &self.launch_hash)
                    || !constant_time_digest_eq(&nonce_sha256, &self.nonce_sha256)
                {
                    return Err(BrokerError::Protocol(
                        "broker completion response did not match its connection".into(),
                    ));
                }
                Ok(record)
            }
            response => response.into_error(),
        }
    }

    fn rpc(&mut self, request: BrokerRequest) -> BrokerResult<BrokerResponse> {
        write_frame(&mut self.stream, &request)?;
        read_frame(&mut self.stream)
    }
}

#[cfg(any(unix, windows))]
trait BrokerTransport {
    fn peer_identity(&self) -> BrokerResult<ProcessIdentity>;
    fn read_exact(&mut self, bytes: &mut [u8]) -> BrokerResult<()>;
    fn write_all(&mut self, bytes: &[u8]) -> BrokerResult<()>;
}

#[cfg(unix)]
impl BrokerTransport for UnixStream {
    fn peer_identity(&self) -> BrokerResult<ProcessIdentity> {
        peer_identity(self)
    }

    fn read_exact(&mut self, bytes: &mut [u8]) -> BrokerResult<()> {
        std::io::Read::read_exact(self, bytes).map_err(BrokerError::Io)
    }

    fn write_all(&mut self, bytes: &[u8]) -> BrokerResult<()> {
        std::io::Write::write_all(self, bytes).map_err(BrokerError::Io)
    }
}

#[cfg(windows)]
#[allow(unsafe_code)]
impl BrokerTransport for WindowsPipeStream {
    fn peer_identity(&self) -> BrokerResult<ProcessIdentity> {
        if self.client_pid == 0 {
            return Err(BrokerError::PeerCredentialsUnavailable);
        }
        ProcessIdentity::for_pid(self.client_pid)
    }

    fn read_exact(&mut self, bytes: &mut [u8]) -> BrokerResult<()> {
        use windows_sys::Win32::Storage::FileSystem::ReadFile;
        let mut offset = 0;
        while offset < bytes.len() {
            let remaining = bytes.len() - offset;
            let request = u32::try_from(remaining).map_err(|_| {
                BrokerError::Protocol("Windows broker read exceeds u32 size".into())
            })?;
            let mut read = 0_u32;
            let ok = unsafe {
                ReadFile(
                    self.handle,
                    bytes[offset..].as_mut_ptr(),
                    request,
                    &mut read,
                    std::ptr::null_mut(),
                )
            };
            if ok == 0 {
                return Err(BrokerError::Io(io::Error::last_os_error()));
            }
            if read == 0 {
                return Err(BrokerError::Io(io::Error::new(
                    io::ErrorKind::UnexpectedEof,
                    "native broker named pipe closed",
                )));
            }
            offset += usize::try_from(read).expect("Windows read size fits usize");
        }
        Ok(())
    }

    fn write_all(&mut self, bytes: &[u8]) -> BrokerResult<()> {
        use windows_sys::Win32::Storage::FileSystem::WriteFile;
        let mut offset = 0;
        while offset < bytes.len() {
            let remaining = bytes.len() - offset;
            let request = u32::try_from(remaining).map_err(|_| {
                BrokerError::Protocol("Windows broker write exceeds u32 size".into())
            })?;
            let mut written = 0_u32;
            let ok = unsafe {
                WriteFile(
                    self.handle,
                    bytes[offset..].as_ptr(),
                    request,
                    &mut written,
                    std::ptr::null_mut(),
                )
            };
            if ok == 0 {
                return Err(BrokerError::Io(io::Error::last_os_error()));
            }
            if written == 0 {
                return Err(BrokerError::Io(io::Error::new(
                    io::ErrorKind::WriteZero,
                    "native broker named pipe wrote no bytes",
                )));
            }
            offset += usize::try_from(written).expect("Windows write size fits usize");
        }
        Ok(())
    }
}

#[cfg(any(unix, windows))]
#[derive(serde::Deserialize, serde::Serialize)]
#[serde(tag = "op", content = "value", deny_unknown_fields)]
enum BrokerRequest {
    ParentHello {
        schema: String,
        harness: Harness,
        role: Role,
        session_id: String,
        root_session_id: String,
        dispatch_id: Option<String>,
    },
    Prepare {
        request: PreparePendingDispatchRequest,
    },
    RegisterChild {
        launch_id: [u8; 32],
        pid: u32,
    },
    Cleanup {
        launch_id: [u8; 32],
        state: PendingLaunchState,
    },
    ChildHello {
        launch_id: [u8; 32],
    },
    ChildEventHello {
        launch_id: [u8; 32],
    },
    Claim {
        launch_id: [u8; 32],
        agent_id: String,
        session_id: String,
        agent_type: String,
        attestation: LoadedCarrierAttestationV1,
    },
    Activate {
        launch_id: [u8; 32],
        agent_id: String,
        session_id: String,
        agent_type: String,
        attestation: LoadedCarrierAttestationV1,
    },
    Complete {
        launch_id: [u8; 32],
        agent_id: String,
        session_id: String,
        agent_type: String,
        nonce_sha256: [u8; 32],
    },
}

#[cfg(any(unix, windows))]
#[allow(clippy::large_enum_variant)]
#[derive(serde::Deserialize, serde::Serialize)]
#[serde(tag = "kind", content = "value", deny_unknown_fields)]
enum BrokerResponse {
    Ok,
    Prepared {
        launch_id: [u8; 32],
        run: String,
        nonce_sha256: [u8; 32],
    },
    ChildAccepted {
        run: String,
        launch_hash: [u8; 32],
        nonce_sha256: [u8; 32],
        expected_attachment: CarrierAttachmentExpectation,
    },
    Claimed,
    Activated {
        record: shepherd::dispatch::DispatchRecord,
    },
    Completed {
        schema: String,
        launch_id_hash: [u8; 32],
        nonce_sha256: [u8; 32],
        record: DispatchRecord,
    },
    Cleaned {
        response: shepherd::dispatch::LaunchCleanupResponse,
    },
    Error {
        code: String,
        detail: String,
    },
}

#[cfg(any(unix, windows))]
impl BrokerResponse {
    fn into_error<T>(self) -> BrokerResult<T> {
        match self {
            Self::Error { code, detail } => Err(BrokerError::Protocol(format!("{code}: {detail}"))),
            _ => Err(BrokerError::Protocol(
                "unexpected native broker response".into(),
            )),
        }
    }
}

#[cfg(any(unix, windows))]
fn handle_connection<S: BrokerTransport>(
    mut stream: S,
    service: DispatchService,
    state: Arc<Mutex<BrokerState>>,
) {
    let peer = match stream.peer_identity() {
        Ok(peer) => peer,
        Err(_) => return,
    };
    let mut parent: Option<BrokerParent> = None;
    let mut child: Option<BrokerChild> = None;
    while let Ok(request) = read_frame::<BrokerRequest, S>(&mut stream) {
        let response = match request {
            BrokerRequest::ParentHello {
                schema,
                harness,
                role,
                session_id,
                root_session_id,
                dispatch_id,
            } => {
                if parent.is_some() {
                    return send_error(
                        &mut stream,
                        "protocol",
                        "parent registration is one-shot per connection".into(),
                    );
                }
                if schema != BROKER_PROTOCOL_SCHEMA {
                    return send_error(&mut stream, "protocol", "unsupported broker schema".into());
                }
                let session_id = match SessionId::new(session_id) {
                    Ok(value) => value,
                    Err(error) => return send_error(&mut stream, "parent", error.to_string()),
                };
                let root_session_id = match SessionId::new(root_session_id) {
                    Ok(value) => value,
                    Err(error) => return send_error(&mut stream, "parent", error.to_string()),
                };
                let dispatch_id = match dispatch_id
                    .map(DispatchId::new)
                    .transpose()
                    .map_err(|error| error.to_string())
                {
                    Ok(value) => value,
                    Err(error) => return send_error(&mut stream, "parent", error),
                };
                let requested_parent = BrokerParent {
                    process: peer,
                    harness,
                    role,
                    session_id,
                    root_session_id,
                    dispatch_id,
                };
                let authenticated = match state.lock() {
                    Ok(guard) => {
                        if peer.same_process(guard.owner) {
                            Some(requested_parent.clone())
                        } else {
                            guard
                                .launches
                                .values()
                                .filter_map(|launch| launch.active_parent.as_ref())
                                .find(|registered| {
                                    registered.process.same_process(peer)
                                        && registered.harness == requested_parent.harness
                                        && registered.role == requested_parent.role
                                        && registered.session_id == requested_parent.session_id
                                        && registered.root_session_id
                                            == requested_parent.root_session_id
                                        && registered.dispatch_id == requested_parent.dispatch_id
                                })
                                .cloned()
                        }
                    }
                    Err(_) => {
                        return send_error(&mut stream, "broker", "state lock poisoned".into());
                    }
                };
                let Some(authenticated) = authenticated else {
                    return send_error(
                        &mut stream,
                        "wrong-parent",
                        "parent process identity or session binding is not registered".into(),
                    );
                };
                parent = Some(authenticated);
                BrokerResponse::Ok
            }
            BrokerRequest::Prepare { request } => {
                let Some(parent_auth) = parent.as_ref() else {
                    return send_error(
                        &mut stream,
                        "parent-required",
                        "parent registration required".into(),
                    );
                };
                let launch_id = match BrokerLaunchId::fresh() {
                    Ok(value) => value,
                    Err(error) => return send_error(&mut stream, "entropy", error.to_string()),
                };
                let secret = match os_random_32() {
                    Ok(value) => BrokerSecret(value),
                    Err(error) => return send_error(&mut stream, "entropy", error.to_string()),
                };
                match service.prepare_pending(
                    parent_auth,
                    request,
                    launch_id,
                    digest(&secret.0),
                    now_millis(),
                ) {
                    Ok(value) => {
                        let run = value.pending.run.clone();
                        let mut guard = match state.lock() {
                            Ok(guard) => guard,
                            Err(_) => {
                                return send_error(
                                    &mut stream,
                                    "broker",
                                    "state lock poisoned".into(),
                                );
                            }
                        };
                        guard.launches.insert(
                            launch_id,
                            LaunchRecord {
                                secret,
                                parent: parent_auth.clone(),
                                run: run.clone(),
                                nonce_sha256: value.pending.nonce_sha256,
                                expected_attachment: value.pending.expected_attachment.clone(),
                                child: None,
                                child_connected: false,
                                activated: false,
                                completed: false,
                                active_parent: None,
                            },
                        );
                        BrokerResponse::Prepared {
                            launch_id: launch_id.0,
                            run: run.to_string(),
                            nonce_sha256: value.pending.nonce_sha256,
                        }
                    }
                    Err(error) => BrokerResponse::Error {
                        code: "prepare".into(),
                        detail: error.to_string(),
                    },
                }
            }
            BrokerRequest::RegisterChild { launch_id, pid } => {
                let Some(parent_auth) = parent.as_ref() else {
                    return send_error(
                        &mut stream,
                        "parent-required",
                        "parent registration required".into(),
                    );
                };
                let launch_id = BrokerLaunchId::from_bytes(launch_id);
                let mut guard = match state.lock() {
                    Ok(guard) => guard,
                    Err(_) => {
                        return send_error(&mut stream, "broker", "state lock poisoned".into());
                    }
                };
                let Some(launch) = guard.launches.get_mut(&launch_id) else {
                    return send_error(
                        &mut stream,
                        "unknown-launch",
                        "launch identity is unknown".into(),
                    );
                };
                if !launch.parent.process.same_process(parent_auth.process) {
                    return send_error(
                        &mut stream,
                        "wrong-parent",
                        "launch parent identity mismatch".into(),
                    );
                }
                if launch.child.is_some() {
                    return send_error(
                        &mut stream,
                        "replay",
                        "child process registration was already consumed".into(),
                    );
                }
                let child_identity = match ProcessIdentity::for_pid(pid) {
                    Ok(identity) => identity,
                    Err(error) => return send_error(&mut stream, "child", error.to_string()),
                };
                match is_descendant(child_identity, parent_auth.process) {
                    Ok(true) => {
                        launch.child = Some(child_identity);
                        BrokerResponse::Ok
                    }
                    Ok(false) => BrokerResponse::Error {
                        code: "wrong-ancestry".into(),
                        detail: "registered child is not a descendant of the parent".into(),
                    },
                    Err(error) => BrokerResponse::Error {
                        code: "ancestry".into(),
                        detail: error.to_string(),
                    },
                }
            }
            BrokerRequest::Cleanup {
                launch_id,
                state: launch_state,
            } => {
                let Some(parent_auth) = parent.as_ref() else {
                    return send_error(
                        &mut stream,
                        "parent-required",
                        "parent registration required".into(),
                    );
                };
                if !launch_state.is_terminal() {
                    return send_error(
                        &mut stream,
                        "cleanup",
                        "launch cleanup state must be terminal".into(),
                    );
                }
                let launch_id = BrokerLaunchId::from_bytes(launch_id);
                let launch_hash = {
                    let guard = match state.lock() {
                        Ok(guard) => guard,
                        Err(_) => {
                            return send_error(&mut stream, "broker", "state lock poisoned".into());
                        }
                    };
                    let Some(launch) = guard.launches.get(&launch_id) else {
                        return send_error(
                            &mut stream,
                            "unknown-launch",
                            "launch identity is unknown".into(),
                        );
                    };
                    if !launch.parent.process.same_process(parent_auth.process) {
                        return send_error(
                            &mut stream,
                            "wrong-parent",
                            "launch parent identity mismatch".into(),
                        );
                    }
                    if launch.activated {
                        return send_error(
                            &mut stream,
                            "active-launch",
                            "an activated launch cannot be cleaned up by its parent".into(),
                        );
                    }
                    (launch.run.clone(), digest(&launch.secret.0))
                };
                match service
                    .store()
                    .cleanup_pending(&launch_hash.0, launch_hash.1, launch_state)
                {
                    Ok(response) => {
                        if let Ok(mut guard) = state.lock() {
                            guard.launches.remove(&launch_id);
                        }
                        BrokerResponse::Cleaned { response }
                    }
                    Err(error) => BrokerResponse::Error {
                        code: "cleanup".into(),
                        detail: error.to_string(),
                    },
                }
            }
            BrokerRequest::ChildEventHello { launch_id } => {
                let launch_id = BrokerLaunchId::from_bytes(launch_id);
                let mut guard = match state.lock() {
                    Ok(guard) => guard,
                    Err(_) => {
                        return send_error(&mut stream, "broker", "state lock poisoned".into());
                    }
                };
                let Some(launch) = guard.launches.get_mut(&launch_id) else {
                    return send_error(
                        &mut stream,
                        "unknown-launch",
                        "launch identity is unknown".into(),
                    );
                };
                let Some(expected_child) = launch.child else {
                    return send_error(
                        &mut stream,
                        "child-not-registered",
                        "child process was not registered".into(),
                    );
                };
                if !launch.activated || launch.completed {
                    return send_error(
                        &mut stream,
                        "replay",
                        "child event is not valid for this launch state".into(),
                    );
                }
                let hook_is_descendant = match is_descendant(peer, expected_child) {
                    Ok(value) => value,
                    Err(error) => return send_error(&mut stream, "ancestry", error.to_string()),
                };
                if !expected_child.same_process(peer) && !hook_is_descendant {
                    return send_error(
                        &mut stream,
                        "wrong-pid",
                        "event peer is not the registered harness or its descendant hook".into(),
                    );
                }
                child = Some(BrokerChild {
                    process: expected_child,
                    parent_process_hash: launch.parent.process.digest(),
                    launch_id,
                    launch_hash: digest(&launch.secret.0),
                    run: launch.run.clone(),
                    nonce_sha256: launch.nonce_sha256,
                    expected_attachment: launch.expected_attachment.clone(),
                    initial_connection: false,
                });
                BrokerResponse::ChildAccepted {
                    run: launch.run.to_string(),
                    launch_hash: digest(&launch.secret.0),
                    nonce_sha256: launch.nonce_sha256,
                    expected_attachment: launch.expected_attachment.clone(),
                }
            }
            BrokerRequest::ChildHello { launch_id } => {
                let launch_id = BrokerLaunchId::from_bytes(launch_id);
                let mut guard = match state.lock() {
                    Ok(guard) => guard,
                    Err(_) => {
                        return send_error(&mut stream, "broker", "state lock poisoned".into());
                    }
                };
                let Some(launch) = guard.launches.get_mut(&launch_id) else {
                    return send_error(
                        &mut stream,
                        "unknown-launch",
                        "launch identity is unknown".into(),
                    );
                };
                let Some(expected_child) = launch.child else {
                    return send_error(
                        &mut stream,
                        "child-not-registered",
                        "child process was not registered".into(),
                    );
                };
                if launch.child_connected {
                    return send_error(
                        &mut stream,
                        "replay",
                        "child connection was already consumed".into(),
                    );
                }
                let hook_is_descendant = match is_descendant(peer, expected_child) {
                    Ok(value) => value,
                    Err(error) => return send_error(&mut stream, "ancestry", error.to_string()),
                };
                if !expected_child.same_process(peer) && !hook_is_descendant {
                    return send_error(
                        &mut stream,
                        "wrong-pid",
                        "peer is not the registered child or its descendant hook".into(),
                    );
                }
                match is_descendant(peer, launch.parent.process) {
                    Ok(true) => {}
                    Ok(false) if !peer.same_process(launch.parent.process) => {
                        return send_error(
                            &mut stream,
                            "wrong-ancestry",
                            "child is no longer a descendant of the registered parent".into(),
                        );
                    }
                    Ok(false) => {}
                    Err(error) => return send_error(&mut stream, "ancestry", error.to_string()),
                }
                launch.child_connected = true;
                child = Some(BrokerChild {
                    process: expected_child,
                    parent_process_hash: launch.parent.process.digest(),
                    launch_id,
                    launch_hash: digest(&launch.secret.0),
                    run: launch.run.clone(),
                    nonce_sha256: launch.nonce_sha256,
                    expected_attachment: launch.expected_attachment.clone(),
                    initial_connection: true,
                });
                BrokerResponse::ChildAccepted {
                    run: launch.run.to_string(),
                    launch_hash: digest(&launch.secret.0),
                    nonce_sha256: launch.nonce_sha256,
                    expected_attachment: child
                        .as_ref()
                        .expect("child auth just installed")
                        .expected_attachment
                        .clone(),
                }
            }
            BrokerRequest::Claim {
                launch_id,
                agent_id,
                session_id,
                agent_type,
                attestation,
            } => {
                let Some(child_auth) = child.as_ref() else {
                    return send_error(
                        &mut stream,
                        "child-required",
                        "child connection required".into(),
                    );
                };
                let request = ClaimPendingDispatchRequest {
                    schema: "shepherd.pending-dispatch-request/2".into(),
                    launch_id: BrokerLaunchId::from_bytes(launch_id),
                    agent_id,
                    session_id,
                    agent_type,
                    attestation,
                };
                match service.claim_pending(child_auth, request) {
                    Ok(_) => BrokerResponse::Claimed,
                    Err(error) => BrokerResponse::Error {
                        code: "claim".into(),
                        detail: error.to_string(),
                    },
                }
            }
            BrokerRequest::Activate {
                launch_id,
                agent_id,
                session_id,
                agent_type,
                attestation,
            } => {
                let Some(child_auth) = child.as_ref() else {
                    return send_error(
                        &mut stream,
                        "child-required",
                        "child connection required".into(),
                    );
                };
                let request = ClaimPendingDispatchRequest {
                    schema: "shepherd.pending-dispatch-request/2".into(),
                    launch_id: BrokerLaunchId::from_bytes(launch_id),
                    agent_id,
                    session_id,
                    agent_type,
                    attestation,
                };
                match service.activate_pending(child_auth, request) {
                    Ok(record) => {
                        if let Ok(mut guard) = state.lock()
                            && let Some(launch) = guard.launches.get_mut(&child_auth.launch_id)
                        {
                            launch.activated = true;
                            launch.active_parent = Some(BrokerParent {
                                process: child_auth.process,
                                harness: record.harness,
                                role: record.role,
                                session_id: record.session_id.clone(),
                                root_session_id: launch.parent.root_session_id.clone(),
                                dispatch_id: DispatchId::new(record.agent_id.to_string()).ok(),
                            });
                        }
                        BrokerResponse::Activated { record }
                    }
                    Err(error) => BrokerResponse::Error {
                        code: "activate".into(),
                        detail: error.to_string(),
                    },
                }
            }
            BrokerRequest::Complete {
                launch_id,
                agent_id,
                session_id,
                agent_type,
                nonce_sha256,
            } => {
                let Some(child_auth) = child.as_ref() else {
                    return send_error(
                        &mut stream,
                        "child-required",
                        "child event connection required".into(),
                    );
                };
                if child_auth.initial_connection {
                    return send_error(
                        &mut stream,
                        "child-required",
                        "child completion requires a later event connection".into(),
                    );
                }
                if BrokerLaunchId::from_bytes(launch_id) != child_auth.launch_id
                    || !constant_time_digest_eq(&nonce_sha256, &child_auth.nonce_sha256)
                {
                    return send_error(
                        &mut stream,
                        "replay",
                        "child completion does not match its broker connection".into(),
                    );
                }
                match service.complete_broker_child(
                    child_auth,
                    agent_id,
                    session_id,
                    agent_type,
                    nonce_sha256,
                ) {
                    Ok(record) => {
                        if let Ok(mut guard) = state.lock()
                            && let Some(launch) = guard.launches.get_mut(&child_auth.launch_id)
                        {
                            launch.completed = true;
                        }
                        BrokerResponse::Completed {
                            schema: BROKER_COMPLETION_SCHEMA.into(),
                            launch_id_hash: child_auth.launch_hash,
                            nonce_sha256: child_auth.nonce_sha256,
                            record,
                        }
                    }
                    Err(error) => BrokerResponse::Error {
                        code: "complete".into(),
                        detail: error.to_string(),
                    },
                }
            }
        };
        if write_frame(&mut stream, &response).is_err() {
            break;
        }
    }
    if let Some(child_auth) = child {
        let should_cleanup = child_auth.initial_connection
            && state
                .lock()
                .ok()
                .and_then(|guard| {
                    guard
                        .launches
                        .get(&child_auth.launch_id)
                        .map(|launch| !launch.activated)
                })
                .unwrap_or(false);
        if should_cleanup {
            let _ = service.store().cleanup_pending(
                &child_auth.run,
                child_auth.launch_hash,
                PendingLaunchState::LaunchFailed,
            );
            if let Ok(mut guard) = state.lock() {
                guard.launches.remove(&child_auth.launch_id);
            }
        }
    }
}

#[cfg(any(unix, windows))]
fn send_error<S: BrokerTransport>(stream: &mut S, code: &str, detail: String) {
    let _ = write_frame(
        stream,
        &BrokerResponse::Error {
            code: code.into(),
            detail,
        },
    );
}

#[cfg(any(unix, windows))]
pub(crate) fn now_millis() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .ok()
        .and_then(|duration| i64::try_from(duration.as_millis()).ok())
        .unwrap_or(i64::MAX)
}

#[cfg(any(unix, windows))]
fn write_frame<T: serde::Serialize, S: BrokerTransport>(
    stream: &mut S,
    value: &T,
) -> BrokerResult<()> {
    let bytes = serde_json::to_vec(value)
        .map_err(|error| BrokerError::Protocol(format!("cannot encode broker frame: {error}")))?;
    let length = u32::try_from(bytes.len())
        .map_err(|_| BrokerError::Protocol("broker frame is too large".into()))?;
    stream.write_all(&length.to_be_bytes())?;
    stream.write_all(&bytes)?;
    Ok(())
}

#[cfg(any(unix, windows))]
fn read_frame<T: serde::de::DeserializeOwned, S: BrokerTransport>(
    stream: &mut S,
) -> BrokerResult<T> {
    let mut length = [0_u8; 4];
    stream.read_exact(&mut length)?;
    let length = u32::from_be_bytes(length);
    if length == 0 || length > 1_048_576 {
        return Err(BrokerError::Protocol("broker frame size is invalid".into()));
    }
    let mut bytes = vec![0_u8; usize::try_from(length).expect("u32 fits usize")];
    stream.read_exact(&mut bytes)?;
    serde_json::from_slice(&bytes)
        .map_err(|error| BrokerError::Protocol(format!("cannot decode broker frame: {error}")))
}

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

    #[test]
    fn launch_secret_uses_direct_os_csprng_and_never_formats_plaintext() {
        let first = os_random_32().expect("OS CSPRNG");
        let second = os_random_32().expect("OS CSPRNG");
        assert_ne!(first, second);
        let secret = BrokerSecret(first);
        let debug = format!("{secret:?}");
        assert!(!debug.contains(&hex_bytes(&first)));
        assert!(!debug.contains(String::from_utf8_lossy(&first).as_ref()));
    }

    #[test]
    fn process_identity_debug_redacts_start_material() {
        let identity = ProcessIdentity::current().expect("current process identity");
        let debug = format!("{identity:?}");
        assert!(!debug.contains(&hex_bytes(&identity.start_hash)));
        assert_eq!(identity.pid(), std::process::id());
    }
}