use crate::scope::Escalation;
use crate::{self as sys, Child, Scope};
use avril_caps::{Capabilities, PathTarget, env_name};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet};
use std::ffi::{OsStr, OsString};
use std::io;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::process::ExitStatus;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::runtime::Handle;
use tokio::sync::{Mutex as AsyncMutex, Notify};
use tokio_util::sync::CancellationToken;

type Reader = Pin<Box<dyn AsyncRead + Send>>;
type Writer = Pin<Box<dyn AsyncWrite + Send>>;
const CHUNK: usize = 64 * 1024;
fn invalid(message: &str) -> io::Error {
    io::Error::new(io::ErrorKind::InvalidInput, message)
}
fn missing() -> io::Error {
    io::Error::new(io::ErrorKind::NotConnected, "Subprocess handle is closed")
}
fn native_env_name(name: &OsStr) -> OsString {
    if cfg!(windows) {
        name.to_ascii_uppercase()
    } else {
        name.to_owned()
    }
}

#[derive(Default, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct Options {
    pub command: String,
    pub args: Vec<String>,
    pub cwd: Option<PathBuf>,
    pub search_path: bool,
    pub clear_env: bool,
    pub env: Vec<(String, String)>,
    pub stdin: StdioSpec,
    pub stdout: StdioSpec,
    pub stderr: StdioSpec,
    pub extra_stdio: Vec<(u32, StdioSpec)>,
    pub terminal: Option<TerminalSize>,
    pub timeout_ms: Option<u64>,
    /// Caps bytes produced on outputs collected natively, which only runSync
    /// does; JS enforces the collected limit everywhere else.
    pub max_collected_bytes: Option<u64>,
    pub max_buffered_bytes: u64,
    pub termination: Termination,
    pub drain_timeout_ms: u64,
    /// Outputs whose reader keeps up deliberately: at the unread limit the
    /// pump stops reading, so the child blocks instead of being terminated.
    #[serde(default)]
    pub backpressure: Vec<u32>,
}
impl Options {
    pub fn new(command: impl Into<String>, args: Vec<String>) -> Self {
        Self {
            command: command.into(),
            args,
            search_path: true,
            max_buffered_bytes: 1024 * 1024,
            drain_timeout_ms: 1000,
            termination: Termination {
                scope: "group".into(),
                signal: "SIGTERM".into(),
                grace_ms: 1000,
            },
            ..Self::default()
        }
    }
    fn stdio(&self) -> impl Iterator<Item = (u32, &StdioSpec, bool)> {
        let standard = [
            (0, &self.stdin, true),
            (1, &self.stdout, false),
            (2, &self.stderr, false),
        ];
        let extra = self.extra_stdio.iter().map(|(fd, spec)| {
            (
                *fd,
                spec,
                spec.direction.as_deref() == Some("parent-to-child"),
            )
        });
        standard.into_iter().chain(extra)
    }
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct StdioSpec {
    pub mode: String,
    pub file: Option<String>,
    pub file_mode: Option<String>,
    pub handle: Option<u64>,
    pub direction: Option<String>,
}
impl Default for StdioSpec {
    fn default() -> Self {
        Self {
            mode: "inherit".into(),
            file: None,
            file_mode: None,
            handle: None,
            direction: None,
        }
    }
}
#[derive(Deserialize)]
pub struct TerminalSize {
    pub columns: u16,
    pub rows: u16,
}
#[derive(Default, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct Termination {
    pub scope: String,
    pub signal: String,
    pub grace_ms: u64,
}

enum StdioPlan<'a> {
    Inherit,
    Null,
    Pipe,
    Stdout,
    File(&'a StdioSpec),
    Handle(&'a StdioSpec),
}

enum TerminationScope {
    Child,
    Group,
}

struct SpawnPlan<'a> {
    stdio: Vec<(u32, StdioPlan<'a>, bool)>,
    scope: TerminationScope,
    terminal: Option<&'a TerminalSize>,
}
#[derive(Clone, Serialize)]
pub struct Status {
    pub code: Option<i32>,
    pub signal: Option<String>,
    pub success: bool,
    pub reason: Option<String>,
    pub error: Option<String>,
}
#[derive(Default)]
struct Queue {
    bytes: bytes::BytesMut,
    closed: bool,
}
#[derive(Default)]
pub struct Output {
    queue: Mutex<Queue>,
    notify: Notify,
    backpressure: bool,
    /// Signals the pump when a backpressured reader frees queue space.
    space: Notify,
}
impl Output {
    fn new(backpressure: bool) -> Self {
        Self { backpressure, ..Self::default() }
    }
    pub async fn read(&self) -> Option<bytes::Bytes> {
        loop {
            let wake = self.notify.notified();
            tokio::pin!(wake);
            wake.as_mut().enable();
            {
                let mut queue = self.queue.lock().unwrap();
                if !queue.bytes.is_empty() {
                    let count = queue.bytes.len().min(CHUNK);
                    let bytes = queue.bytes.split_to(count).freeze();
                    self.space.notify_one();
                    return Some(bytes);
                }
                if queue.closed {
                    return None;
                }
            }
            wake.await;
        }
    }
    pub fn cancel(&self) {
        let mut queue = self.queue.lock().unwrap();
        queue.closed = true;
        queue.bytes = bytes::BytesMut::new();
        self.notify.notify_waiters();
        self.space.notify_one();
    }
}
pub struct Input {
    writer: AsyncMutex<Option<Writer>>,
    stop: CancellationToken,
}
impl Input {
    pub async fn write(&self, bytes: Vec<u8>) -> io::Result<()> {
        tokio::select! {
            biased;
            _ = self.stop.cancelled() => Err(io::Error::new(io::ErrorKind::BrokenPipe, "Subprocess input is closed")),
            result = async { self.writer.lock().await.as_mut().ok_or_else(missing)?.write_all(&bytes).await } => result,
        }
    }
    pub async fn close(&self) {
        self.stop.cancel();
        self.writer.lock().await.take();
    }
}
struct Core {
    child: Option<Child>,
    exit: Option<ExitStatus>,
    completed: bool,
    reason: Option<String>,
    error: Option<String>,
    stop: Escalation,
    epoch: u32,
    produced: u64,
}
impl Core {
    fn observe(&mut self) -> io::Result<bool> {
        if self.exit.is_some() {
            return Ok(false);
        }
        self.exit = self.child.as_ref().map_or(Ok(None), Child::observe_exit)?;
        Ok(self.exit.is_some())
    }
    fn report(&mut self, reason: &str, error: Option<String>) {
        if self.reason.is_none() {
            self.reason = Some(reason.into());
            self.error = error;
        }
    }
}
pub struct Process {
    pub pid: u32,
    group: bool,
    drain: Duration,
    timeout: Option<Duration>,
    limit: Option<u64>,
    pending_limit: u64,
    core: Mutex<Core>,
    done: Condvar,
    changed: Notify,
    wake: Notify,
    pipe_stop: CancellationToken,
    pub outputs: BTreeMap<u32, Arc<Output>>,
    pub inputs: BTreeMap<u32, Arc<Input>>,
    pumps: AtomicUsize,
    /// Pumps waiting for a backpressured reader to free space.
    parked: AtomicUsize,
    owners: Vec<CancellationToken>,
    window: Option<sys::Window>,
}
impl Process {
    pub fn wait_blocking(&self, timeout: Duration) {
        let core = self.core.lock().unwrap();
        let _ = self
            .done
            .wait_timeout_while(core, timeout, |core| !core.completed);
    }
    pub fn invalidate_waiters(&self) -> u32 {
        let mut core = self.core.lock().unwrap();
        core.epoch = core.epoch.wrapping_add(1);
        self.changed.notify_waiters();
        core.epoch
    }
    pub fn resize(&self, columns: u16, rows: u16) -> io::Result<()> {
        sys::resize(
            self.window
                .as_ref()
                .ok_or_else(|| invalid("Subprocess has no terminal"))?,
            columns,
            rows,
        )
    }

    fn status(core: &Core) -> Option<Status> {
        let exit = core.exit?;
        Some(Status {
            code: exit.code(),
            signal: sys::exit_signal(&exit),
            success: exit.success() && core.reason.is_none(),
            reason: core.reason.clone(),
            error: core.error.clone(),
        })
    }
    pub fn request(&self, reason: &str, error: Option<String>, force: bool) {
        let mut core = self.core.lock().unwrap();
        if core.completed {
            return;
        }
        // Supervision polls for owner shutdown, so repeat requests must not
        // re-notify and spin the supervisor into starving pipe tasks.
        core.report(reason, error);
        let changed = core.stop.request(force);
        if changed {
            self.wake.notify_one();
        }
    }
    /// Synchronously SIGKILLs the termination scope. Used on the dying path
    /// so grace-resisting descendants cannot outlive the owner.
    pub fn kill_scope(&self) {
        let mut core = self.core.lock().unwrap();
        let _ = self.signal_locked(&mut core, "SIGKILL");
    }
    fn signal_locked(&self, core: &mut Core, signal: &str) -> io::Result<bool> {
        match &core.child {
            Some(child) => child.signal(self.group, signal),
            None => Ok(false),
        }
    }
    pub fn kill(&self, signal: &str) -> io::Result<bool> {
        let mut core = self.core.lock().unwrap();
        if core.exit.is_some() || core.completed {
            return Ok(false);
        }
        if core.observe()? {
            self.changed.notify_waiters();
            return Ok(false);
        }
        self.signal_locked(&mut core, signal)
    }
    pub async fn wait(&self, completed: bool, epoch: Option<u32>) -> io::Result<Option<Status>> {
        loop {
            let wake = self.changed.notified();
            tokio::pin!(wake);
            wake.as_mut().enable();
            {
                let core = self.core.lock().unwrap();
                if epoch.is_some_and(|epoch| epoch != core.epoch) {
                    return Ok(None);
                }
                if (!completed && core.exit.is_some()) || core.completed {
                    return Ok(Self::status(&core));
                }
            }
            wake.await;
        }
    }
    /// Waits while a backpressured output is full; false once pipes stop.
    async fn room(&self, output: &Output) -> bool {
        loop {
            let space = output.space.notified();
            tokio::pin!(space);
            space.as_mut().enable();
            {
                let queue = output.queue.lock().unwrap();
                if queue.closed || (queue.bytes.len() as u64) < self.pending_limit {
                    return true;
                }
            }
            self.parked.fetch_add(1, Ordering::AcqRel);
            self.wake.notify_one();
            let stopped = tokio::select! {
                biased;
                _ = self.pipe_stop.cancelled() => true,
                _ = space => false,
            };
            self.parked.fetch_sub(1, Ordering::AcqRel);
            self.wake.notify_one();
            if stopped {
                return false;
            }
        }
    }
    async fn pump(self: Arc<Self>, output: Arc<Output>, mut reader: Reader) {
        let mut bytes = vec![0; CHUNK];
        loop {
            if output.backpressure && !self.room(&output).await {
                break;
            }
            let count = tokio::select! { biased; _ = self.pipe_stop.cancelled() => break,
            result = reader.read(&mut bytes) => match result { Ok(0) => break, Ok(count) => count,
                Err(error) => { self.request("io-error", Some(error.to_string()), false); break; } } };
            let overflow;
            {
                let mut core = self.core.lock().unwrap();
                let mut queue = output.queue.lock().unwrap();
                if core.reason.as_deref() == Some("output-limit") {
                    continue;
                }
                let allowed = self
                    .limit
                    .map_or(count as u64, |limit| limit.saturating_sub(core.produced))
                    .min(count as u64);
                core.produced = core.produced.saturating_add(count as u64);
                let allowed = if queue.closed || output.backpressure {
                    allowed
                } else {
                    allowed.min(self.pending_limit.saturating_sub(queue.bytes.len() as u64))
                };
                overflow = allowed < count as u64;
                if !queue.closed && allowed > 0 {
                    queue.bytes.extend_from_slice(&bytes[..allowed as usize]);
                }
            }
            output.notify.notify_waiters();
            if overflow {
                self.request(
                    "output-limit",
                    Some("Subprocess output exceeded its configured limit".into()),
                    false,
                );
            }
        }
        output.queue.lock().unwrap().closed = true;
        output.notify.notify_waiters();
        self.pumps.fetch_sub(1, Ordering::AcqRel);
        self.wake.notify_one();
    }
    async fn supervise(self: Arc<Self>) {
        let started = Instant::now();
        let mut drain_from = None;
        loop {
            if self.owners.iter().any(CancellationToken::is_cancelled) {
                self.request("owner", None, true);
            }
            {
                let mut core = self.core.lock().unwrap();
                match core.observe() {
                    Ok(true) => self.changed.notify_waiters(),
                    Ok(false) => (),
                    Err(error) if error.kind() == io::ErrorKind::Interrupted => (),
                    Err(error) => {
                        // waitid only fails persistently when the child is already
                        // gone (ECHILD) or the arguments are invalid; either way no
                        // exit will ever be observed, so drain pipes and finish
                        // instead of spinning forever.
                        core.report("io-error", Some(error.to_string()));
                        core.stop.request(false);
                        drain_from.get_or_insert_with(Instant::now);
                    }
                }
                if core.exit.is_some() {
                    drain_from.get_or_insert_with(Instant::now);
                }
                if core.reason.is_none()
                    && core.exit.is_none()
                    && self.timeout.is_some_and(|limit| started.elapsed() >= limit)
                {
                    core.reason = Some("timeout".into());
                    core.stop.request(false);
                }
                let Core { child, stop, .. } = &mut *core;
                if let Err(error) = stop.advance(|signal| {
                    child
                        .as_ref()
                        .map_or(Ok(false), |child| child.signal(self.group, signal))
                }) {
                    core.error.get_or_insert(error.to_string());
                }
                if let Some(since) = &mut drain_from {
                    let pumps = self.pumps.load(Ordering::Acquire);
                    // A natural run's pumps waiting on a slow reader are not
                    // pipes held open by stragglers; the drain clock restarts
                    // once they move again.
                    if core.reason.is_none()
                        && pumps != 0
                        && self.parked.load(Ordering::Acquire) == pumps
                    {
                        *since = Instant::now();
                    }
                    // Keep the leader unreaped until group cleanup, so its PID cannot be reused.
                    if since.elapsed() >= self.drain && pumps != 0 {
                        if self.group {
                            let _ = self.signal_locked(&mut core, "SIGKILL");
                        }
                        core.report(
                            "drain-timeout",
                            Some("Subprocess pipes remained open after exit".into()),
                        );
                        self.pipe_stop.cancel();
                    }
                    if self.pumps.load(Ordering::Acquire) == 0 {
                        if self.group {
                            let _ = self.signal_locked(&mut core, "SIGKILL");
                        }
                        break;
                    }
                }
                if core.stop.forced {
                    self.pipe_stop.cancel();
                }
            }
            if drain_from.is_some() {
                for input in self.inputs.values() {
                    input.close().await;
                }
            }
            tokio::select! { _ = self.wake.notified() => (), _ = tokio::time::sleep(Duration::from_millis(5)) => () }
        }
        for input in self.inputs.values() {
            input.close().await;
        }
        let mut core = self.core.lock().unwrap();
        if let Some(mut child) = core.child.take()
            && let Err(error) = child.wait()
        {
            core.error.get_or_insert(error.to_string());
        }
        core.completed = true;
        self.changed.notify_waiters();
        self.done.notify_all();
    }
}

pub struct Launcher {
    pub runtime: Handle,
    pub caps: Arc<Capabilities>,
    pub owners: Vec<CancellationToken>,
}
impl Launcher {
    pub fn spawn(
        &self,
        options: Options,
        environment: impl IntoIterator<Item = (OsString, OsString)>,
        scope: Option<&Scope>,
        file: impl Fn(&StdioSpec, bool, &Path) -> io::Result<std::fs::File>,
    ) -> io::Result<Arc<Process>> {
        let plan = validate(&options)?;
        if self.owners.iter().any(CancellationToken::is_cancelled) {
            return Err(missing());
        }
        let _guard = self.runtime.enter();
        let cwd = options.cwd.as_ref().map_or_else(
            || self.caps.cwd().to_owned(),
            |cwd| self.caps.cwd().join(cwd),
        );
        // Restricted denials share one message so success and failure cannot
        // be used to probe whether directories exist.
        let cwd = cwd.canonicalize().map_err(|error| {
            if self.caps.unrestricted_run() {
                error
            } else {
                denied()
            }
        })?;
        let mut env = BTreeMap::new();
        if !options.clear_env {
            for (name, value) in environment {
                if self.caps.unrestricted_env()
                    || name
                        .to_str()
                        .is_some_and(|name| self.caps.check_env(name).is_ok())
                {
                    env.insert(native_env_name(&name), (name, value));
                }
            }
        }
        for (name, value) in &options.env {
            env.insert(
                native_env_name(OsStr::new(name)),
                (name.into(), value.into()),
            );
        }
        let path = env
            .get(&native_env_name(OsStr::new("PATH")))
            .map(|(_, value)| value.as_os_str());
        let (executable, target) = resolve(
            &options.command,
            options.search_path,
            &cwd,
            path,
            &self.caps,
        )?;
        // A child that inherits standard streams joins the parent's process
        // group instead of its own, but only with a terminal attached: only
        // the foreground group can use it (inherited TTYs, or /dev/tty
        // opened by the child), and seizing the foreground here would
        // suspend this process when restoring it. Headless setups keep
        // their own group so timeouts and kills reach all descendants.
        // Windows consoles deliver to every attached process regardless,
        // so job objects never need to give way.
        let shares_terminal = cfg!(unix)
            && matches!(plan.scope, TerminationScope::Group)
            && plan.terminal.is_none()
            && {
                use std::io::IsTerminal;
                // Options::stdio() yields fds 0, 1, 2 first and in order,
                // so the leading plan entries are the standard streams.
                let inherited = [
                    matches!(plan.stdio[0].1, StdioPlan::Inherit),
                    matches!(plan.stdio[1].1, StdioPlan::Inherit),
                    matches!(plan.stdio[2].1, StdioPlan::Inherit),
                ];
                let terminals = [
                    std::io::stdin().is_terminal(),
                    std::io::stdout().is_terminal(),
                    std::io::stderr().is_terminal(),
                ];
                inherited.contains(&true)
                    && (inherited
                        .into_iter()
                        .zip(terminals)
                        .any(|(inherit, tty)| inherit && tty)
                        || sys::controlling_foreground())
            };
        let group =
            scope.is_none() && matches!(plan.scope, TerminationScope::Group) && !shares_terminal;
        let mut readers: BTreeMap<u32, Reader> = BTreeMap::new();
        let mut writers: BTreeMap<u32, Writer> = BTreeMap::new();
        let mut endpoints = BTreeMap::new();
        let window = if let Some(size) = plan.terminal {
            let pty = sys::pty(size.columns, size.rows)?;
            readers.insert(1, Box::pin(pty.reader));
            writers.insert(0, Box::pin(pty.writer));
            for (fd, endpoint) in pty.endpoints.into_iter().enumerate() {
                endpoints.insert(fd as u32, endpoint);
            }
            Some(pty.window)
        } else {
            for (fd, kind, read) in &plan.stdio {
                let endpoint = match kind {
                    StdioPlan::Pipe => {
                        let (pipe, endpoint) = sys::pipe(*read)?;
                        if *read {
                            writers.insert(*fd, Box::pin(pipe));
                        } else {
                            readers.insert(*fd, Box::pin(pipe));
                        }
                        endpoint
                    }
                    StdioPlan::Null => sys::null(*read)?,
                    StdioPlan::Inherit => sys::inherit(*fd)?,
                    StdioPlan::Stdout => sys::duplicate(&endpoints[&1])?,
                    StdioPlan::File(spec) | StdioPlan::Handle(spec) => {
                        file(spec, *read, &cwd)?.into()
                    }
                };
                endpoints.insert(*fd, endpoint);
            }
            None
        };
        if options
            .backpressure
            .iter()
            .any(|fd| !readers.contains_key(fd))
        {
            return Err(invalid("Backpressure requires a piped output"));
        }
        self.caps.check_run(&target).map_err(|_| denied())?;
        if self.owners.iter().any(CancellationToken::is_cancelled) {
            return Err(missing());
        }
        let launch = sys::Launch {
            executable,
            program: target.as_path(),
            args: &options.args,
            env: env.into_values().collect(),
            cwd: &cwd,
            group,
            join: None,
            terminal: window.is_some(),
            endpoints: endpoints.into_iter().collect(),
        };
        let child = match &scope {
            Some(scope) => scope.spawn(launch)?,
            None => sys::spawn(launch)?,
        };
        let pid = child.id();
        let outputs: BTreeMap<_, _> = readers
            .keys()
            .map(|fd| {
                (
                    *fd,
                    Arc::new(Output::new(options.backpressure.contains(fd))),
                )
            })
            .collect();
        let inputs = writers
            .into_iter()
            .map(|(fd, writer)| {
                (
                    fd,
                    Arc::new(Input {
                        writer: AsyncMutex::new(Some(writer)),
                        stop: CancellationToken::new(),
                    }),
                )
            })
            .collect();
        let process = Arc::new(Process {
            group,
            drain: Duration::from_millis(options.drain_timeout_ms),
            timeout: options.timeout_ms.map(Duration::from_millis),
            limit: options.max_collected_bytes,
            pending_limit: options.max_buffered_bytes,
            core: Mutex::new(Core {
                child: Some(child),
                exit: None,
                completed: false,
                reason: None,
                error: None,
                stop: Escalation::new(vec![(
                    options.termination.signal,
                    Duration::from_millis(options.termination.grace_ms),
                )]),
                epoch: 0,
                produced: 0,
            }),
            done: Condvar::new(),
            changed: Notify::new(),
            wake: Notify::new(),
            pipe_stop: CancellationToken::new(),
            pumps: AtomicUsize::new(readers.len()),
            parked: AtomicUsize::new(0),
            owners: self.owners.clone(),
            pid,
            outputs,
            inputs,
            window,
        });
        for (fd, reader) in readers {
            self.runtime
                .spawn(process.clone().pump(process.outputs[&fd].clone(), reader));
        }
        self.runtime.spawn(process.clone().supervise());
        Ok(process)
    }
}
fn validate(options: &Options) -> io::Result<SpawnPlan<'_>> {
    if options.command.is_empty()
        || options.command.contains('\0')
        || options.args.iter().any(|arg| arg.contains('\0'))
    {
        return Err(invalid("Commands and arguments must not contain NUL"));
    }
    if options.cwd.as_ref().is_some_and(|cwd| {
        cwd.as_os_str().is_empty() || cwd.as_os_str().as_encoded_bytes().contains(&0)
    }) {
        return Err(invalid("Invalid subprocess cwd"));
    }
    for (name, value) in &options.env {
        env_name(name).map_err(|e| invalid(&e.to_string()))?;
        if value.contains('\0') {
            return Err(invalid("Environment values must not contain NUL"));
        }
    }
    if options.max_buffered_bytes == 0
        || options.max_buffered_bytes > (1 << 53) - 1
        || options
            .max_collected_bytes
            .is_some_and(|limit| limit == 0 || limit > (1 << 53) - 1)
    {
        return Err(invalid("Invalid subprocess output limit"));
    }
    let scope = match options.termination.scope.as_str() {
        "group" => TerminationScope::Group,
        "child" => TerminationScope::Child,
        _ => return Err(invalid("Invalid termination scope")),
    };
    sys::validate_signal(&options.termination.signal)?;
    fn validate_spec(
        spec: &StdioSpec,
        read: bool,
        stderr: bool,
        extra: bool,
    ) -> io::Result<StdioPlan<'_>> {
        let plan = match spec.mode.as_str() {
            "inherit" => StdioPlan::Inherit,
            "null" => StdioPlan::Null,
            "piped" => StdioPlan::Pipe,
            "stdout" if stderr => StdioPlan::Stdout,
            "file" => StdioPlan::File(spec),
            "handle" => StdioPlan::Handle(spec),
            _ => return Err(invalid("Invalid stdio mode")),
        };
        if matches!(plan, StdioPlan::File(_)) {
            if spec
                .file
                .as_ref()
                .is_none_or(|path| path.is_empty() || path.contains('\0'))
                || spec.handle.is_some()
            {
                return Err(invalid("Invalid stdio file"));
            }
            if read {
                if spec.file_mode.is_some() {
                    return Err(invalid("Input file accepts no file mode"));
                }
            } else if !matches!(
                spec.file_mode.as_deref(),
                Some("append" | "truncate" | "create-new")
            ) {
                return Err(invalid("Invalid output file mode"));
            }
        } else if spec.file.is_some()
            || spec.file_mode.is_some()
            || matches!(plan, StdioPlan::Handle(_)) != spec.handle.is_some()
        {
            return Err(invalid("Invalid stdio specification"));
        }
        if extra {
            if matches!(plan, StdioPlan::Inherit) {
                return Err(invalid("Extra stdio cannot inherit internal descriptors"));
            }
            if !matches!(
                spec.direction.as_deref(),
                Some("parent-to-child" | "child-to-parent")
            ) {
                return Err(invalid("Extra stdio requires a direction"));
            }
        } else if spec.direction.is_some() {
            return Err(invalid("Unexpected stdio direction"));
        }
        Ok(plan)
    }
    let mut descriptors = BTreeSet::new();
    let mut stdio = Vec::new();
    for (fd, spec, read) in options.stdio() {
        if fd > sys::MAX_DESCRIPTOR || !descriptors.insert(fd) {
            return Err(invalid("Invalid additional descriptor"));
        }
        let plan = validate_spec(spec, read, fd == 2, fd >= 3)?;
        stdio.push((fd, plan, read));
    }
    if let Some(window) = &options.terminal
        && (window.columns == 0
            || window.rows == 0
            || !matches!(stdio[0].1, StdioPlan::Pipe)
            || !matches!(stdio[1].1, StdioPlan::Pipe)
            || !matches!(stdio[2].1, StdioPlan::Stdout)
            || !options.extra_stdio.is_empty())
    {
        return Err(invalid("Terminal stdio must be piped and merged"));
    }
    Ok(SpawnPlan {
        stdio,
        scope,
        terminal: options.terminal.as_ref(),
    })
}

/// Restricted launch denials share one message so they cannot be used to
/// probe whether directories or executables exist.
fn denied() -> io::Error {
    io::Error::new(
        io::ErrorKind::PermissionDenied,
        "Executable is not authorized",
    )
}

fn resolve(
    command: &str,
    search: bool,
    cwd: &Path,
    path: Option<&OsStr>,
    caps: &Capabilities,
) -> io::Result<(sys::Executable, PathTarget)> {
    // Entry-based authorization: the parent is canonicalized but the final
    // component stays lexical, so one granted multi-call entry (busybox,
    // coreutils) does not authorize its siblings.
    let bare = Path::new(command).components().count() == 1
        && !command.contains('/')
        && !command.contains('\\');
    let candidates = if search && bare {
        // An absent PATH under restricted authority must not become an oracle.
        let path = path.ok_or_else(|| {
            if caps.unrestricted_run() {
                io::Error::new(
                    io::ErrorKind::NotFound,
                    "PATH is absent from the subprocess environment",
                )
            } else {
                denied()
            }
        })?;
        std::env::split_paths(path)
            .flat_map(|directory| sys::expand(cwd.join(directory).join(command)))
            .collect::<Vec<_>>()
    } else {
        let candidate = cwd.join(command);
        vec![
            sys::expand(candidate.clone())
                .into_iter()
                .find(|path| path.symlink_metadata().is_ok())
                .unwrap_or(candidate),
        ]
    };
    for candidate in candidates {
        let target = match PathTarget::for_executable(&candidate) {
            Ok(target) => target,
            Err(error)
                if search
                    && bare
                    && matches!(
                        error.kind(),
                        io::ErrorKind::NotFound | io::ErrorKind::NotADirectory
                    ) =>
            {
                continue;
            }
            Err(error) => {
                let permitted = PathTarget::for_creation(&candidate)
                    .ok()
                    .is_some_and(|target| caps.check_run(&target).is_ok());
                return Err(if permitted { error } else { denied() });
            }
        };
        if !(search && bare) {
            caps.check_run(&target).map_err(|_| denied())?;
        }
        let metadata = match std::fs::metadata(target.as_path()) {
            Ok(metadata) => metadata,
            Err(error)
                if search
                    && bare
                    && matches!(
                        error.kind(),
                        io::ErrorKind::NotFound
                            | io::ErrorKind::NotADirectory
                            | io::ErrorKind::PermissionDenied
                    ) =>
            {
                continue;
            }
            Err(error) => return Err(error),
        };
        let executable = sys::is_executable(&metadata);
        if !executable && search && bare {
            continue;
        }
        caps.check_run(&target).map_err(|_| denied())?;
        #[cfg(target_os = "linux")]
        let (executable, metadata) = {
            let (directory, name) = avril_caps::fs::parent(target.as_path())?;
            sys::Executable::open_at(&directory, &name)?
        };
        #[cfg(not(target_os = "linux"))]
        let (executable, metadata) = sys::Executable::open(target.as_path())?;
        if PathTarget::for_executable(&candidate)?.as_path() != target.as_path() {
            return Err(io::Error::other("Executable changed during resolution"));
        }
        if !sys::is_executable(&metadata) {
            return Err(io::Error::new(
                io::ErrorKind::PermissionDenied,
                "Not an executable file",
            ));
        }
        return Ok((executable, target));
    }
    if caps.unrestricted_run() {
        Err(io::Error::new(
            io::ErrorKind::NotFound,
            "Executable was not found",
        ))
    } else {
        Err(denied())
    }
}