rename

since this doesn't just run pipelines anymore

5 files changed, 1466 insertions, 0 deletions

diff --git a/src/runner/builtins/command.rs b/src/runner/builtins/command.rs
new file mode 100644
index 0000000..2c1f755
--- /dev/null
+++ b/src/runner/builtins/command.rs
@@ -0,0 +1,398 @@
+use crate::runner::prelude::*;
+
+use async_std::io::prelude::BufReadExt as _;
+
+pub struct Command {
+    exe: crate::parse::Exe,
+    f: super::Builtin,
+    cfg: Cfg,
+}
+
+impl Command {
+    pub fn new(exe: crate::parse::Exe) -> Result<Self, crate::parse::Exe> {
+        if let Some(s) = exe.exe().to_str() {
+            if let Some(f) = super::BUILTINS.get(s) {
+                Ok(Self {
+                    exe,
+                    f,
+                    cfg: Cfg::new(),
+                })
+            } else {
+                Err(exe)
+            }
+        } else {
+            Err(exe)
+        }
+    }
+
+    pub fn new_with_io(
+        exe: crate::parse::Exe,
+        io: Io,
+    ) -> Result<Self, crate::parse::Exe> {
+        if let Some(s) = exe.exe().to_str() {
+            if let Some(f) = super::BUILTINS.get(s) {
+                Ok(Self {
+                    exe,
+                    f,
+                    cfg: Cfg::new_with_io(io),
+                })
+            } else {
+                Err(exe)
+            }
+        } else {
+            Err(exe)
+        }
+    }
+
+    pub fn stdin(&mut self, fh: std::fs::File) {
+        self.cfg.io.set_stdin(fh);
+    }
+
+    pub fn stdout(&mut self, fh: std::fs::File) {
+        self.cfg.io.set_stdout(fh);
+    }
+
+    pub fn stderr(&mut self, fh: std::fs::File) {
+        self.cfg.io.set_stderr(fh);
+    }
+
+    // Safety: see pre_exec in async_std::os::unix::process::CommandExt (this
+    // is just a wrapper)
+    pub unsafe fn pre_exec<F>(&mut self, f: F)
+    where
+        F: 'static + FnMut() -> std::io::Result<()> + Send + Sync,
+    {
+        self.cfg.pre_exec(f);
+    }
+
+    pub fn apply_redirects(&mut self, redirects: &[crate::parse::Redirect]) {
+        self.cfg.io.apply_redirects(redirects);
+    }
+
+    pub fn spawn(self, env: &Env) -> anyhow::Result<Child> {
+        let Self { f, exe, cfg } = self;
+        (f)(exe, env, cfg)
+    }
+}
+
+pub struct Cfg {
+    io: Io,
+    pre_exec: Option<
+        Box<dyn 'static + FnMut() -> std::io::Result<()> + Send + Sync>,
+    >,
+}
+
+impl Cfg {
+    fn new() -> Self {
+        Self {
+            io: Io::new(),
+            pre_exec: None,
+        }
+    }
+
+    fn new_with_io(io: Io) -> Self {
+        Self { io, pre_exec: None }
+    }
+
+    pub fn io(&self) -> &Io {
+        &self.io
+    }
+
+    // Safety: see pre_exec in async_std::os::unix::process::CommandExt (this
+    // is just a wrapper)
+    pub unsafe fn pre_exec<F>(&mut self, f: F)
+    where
+        F: 'static + FnMut() -> std::io::Result<()> + Send + Sync,
+    {
+        self.pre_exec = Some(Box::new(f));
+    }
+
+    pub fn setup_command(mut self, cmd: &mut crate::runner::Command) {
+        self.io.setup_command(cmd);
+        if let Some(pre_exec) = self.pre_exec.take() {
+            // Safety: pre_exec can only have been set by calling the pre_exec
+            // method, which is itself unsafe, so the safety comments at the
+            // point where that is called are the relevant ones
+            unsafe { cmd.pre_exec(pre_exec) };
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct Io {
+    fds: std::collections::HashMap<
+        std::os::unix::io::RawFd,
+        crate::mutex::Mutex<File>,
+    >,
+}
+
+impl Io {
+    pub fn new() -> Self {
+        Self {
+            fds: std::collections::HashMap::new(),
+        }
+    }
+
+    fn stdin(&self) -> Option<crate::mutex::Mutex<File>> {
+        self.fds.get(&0).map(crate::mutex::clone)
+    }
+
+    pub fn set_stdin<T: std::os::unix::io::IntoRawFd>(&mut self, stdin: T) {
+        if let Some(file) = self.fds.remove(&0) {
+            File::maybe_drop(file);
+        }
+        self.fds.insert(
+            0,
+            // Safety: we just acquired stdin via into_raw_fd, which acquires
+            // ownership of the fd, so we are now the sole owner
+            crate::mutex::new(unsafe { File::input(stdin.into_raw_fd()) }),
+        );
+    }
+
+    fn stdout(&self) -> Option<crate::mutex::Mutex<File>> {
+        self.fds.get(&1).map(crate::mutex::clone)
+    }
+
+    pub fn set_stdout<T: std::os::unix::io::IntoRawFd>(&mut self, stdout: T) {
+        if let Some(file) = self.fds.remove(&1) {
+            File::maybe_drop(file);
+        }
+        self.fds.insert(
+            1,
+            // Safety: we just acquired stdout via into_raw_fd, which acquires
+            // ownership of the fd, so we are now the sole owner
+            crate::mutex::new(unsafe { File::output(stdout.into_raw_fd()) }),
+        );
+    }
+
+    fn stderr(&self) -> Option<crate::mutex::Mutex<File>> {
+        self.fds.get(&2).map(crate::mutex::clone)
+    }
+
+    pub fn set_stderr<T: std::os::unix::io::IntoRawFd>(&mut self, stderr: T) {
+        if let Some(file) = self.fds.remove(&2) {
+            File::maybe_drop(file);
+        }
+        self.fds.insert(
+            2,
+            // Safety: we just acquired stderr via into_raw_fd, which acquires
+            // ownership of the fd, so we are now the sole owner
+            crate::mutex::new(unsafe { File::output(stderr.into_raw_fd()) }),
+        );
+    }
+
+    pub fn apply_redirects(&mut self, redirects: &[crate::parse::Redirect]) {
+        for redirect in redirects {
+            let to = match &redirect.to {
+                crate::parse::RedirectTarget::Fd(fd) => {
+                    crate::mutex::clone(&self.fds[fd])
+                }
+                crate::parse::RedirectTarget::File(path) => {
+                    let fd = redirect.dir.open(path).unwrap();
+                    match redirect.dir {
+                        crate::parse::Direction::In => {
+                            // Safety: we just opened fd, and nothing else has
+                            // or can use it
+                            crate::mutex::new(unsafe { File::input(fd) })
+                        }
+                        crate::parse::Direction::Out
+                        | crate::parse::Direction::Append => {
+                            // Safety: we just opened fd, and nothing else has
+                            // or can use it
+                            crate::mutex::new(unsafe { File::output(fd) })
+                        }
+                    }
+                }
+            };
+            self.fds.insert(redirect.from, to);
+        }
+    }
+
+    pub async fn read_line_stdin(&self) -> anyhow::Result<String> {
+        let mut buf = String::new();
+        if let Some(fh) = self.stdin() {
+            if let File::In(fh) = &mut *fh.lock_arc().await {
+                fh.read_line(&mut buf).await?;
+            }
+        }
+        if buf.ends_with('\n') {
+            buf.truncate(buf.len() - 1);
+        }
+        Ok(buf)
+    }
+
+    pub async fn write_stdout(&self, buf: &[u8]) -> anyhow::Result<()> {
+        if let Some(fh) = self.stdout() {
+            if let File::Out(fh) = &mut *fh.lock_arc().await {
+                Ok(fh.write_all(buf).await.map(|_| ())?)
+            } else {
+                Ok(())
+            }
+        } else {
+            Ok(())
+        }
+    }
+
+    pub async fn write_stderr(&self, buf: &[u8]) -> anyhow::Result<()> {
+        if let Some(fh) = self.stderr() {
+            if let File::Out(fh) = &mut *fh.lock_arc().await {
+                Ok(fh.write_all(buf).await.map(|_| ())?)
+            } else {
+                Ok(())
+            }
+        } else {
+            Ok(())
+        }
+    }
+
+    pub fn setup_command(mut self, cmd: &mut crate::runner::Command) {
+        if let Some(stdin) = self.fds.remove(&0) {
+            if let Some(stdin) = crate::mutex::unwrap(stdin) {
+                let stdin = stdin.into_raw_fd();
+                if stdin != 0 {
+                    // Safety: we just acquired stdin via into_raw_fd, which
+                    // acquires ownership of the fd, so we are now the sole
+                    // owner
+                    cmd.stdin(unsafe { std::fs::File::from_raw_fd(stdin) });
+                    self.fds.remove(&0);
+                }
+            }
+        }
+        if let Some(stdout) = self.fds.remove(&1) {
+            if let Some(stdout) = crate::mutex::unwrap(stdout) {
+                let stdout = stdout.into_raw_fd();
+                if stdout != 1 {
+                    // Safety: we just acquired stdout via into_raw_fd, which
+                    // acquires ownership of the fd, so we are now the sole
+                    // owner
+                    cmd.stdout(unsafe { std::fs::File::from_raw_fd(stdout) });
+                    self.fds.remove(&1);
+                }
+            }
+        }
+        if let Some(stderr) = self.fds.remove(&2) {
+            if let Some(stderr) = crate::mutex::unwrap(stderr) {
+                let stderr = stderr.into_raw_fd();
+                if stderr != 2 {
+                    // Safety: we just acquired stderr via into_raw_fd, which
+                    // acquires ownership of the fd, so we are now the sole
+                    // owner
+                    cmd.stderr(unsafe { std::fs::File::from_raw_fd(stderr) });
+                    self.fds.remove(&2);
+                }
+            }
+        }
+    }
+}
+
+impl Drop for Io {
+    fn drop(&mut self) {
+        for (_, file) in self.fds.drain() {
+            File::maybe_drop(file);
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum File {
+    In(async_std::io::BufReader<async_std::fs::File>),
+    Out(async_std::fs::File),
+}
+
+impl File {
+    // Safety: fd must not be owned by any other File object
+    pub unsafe fn input(fd: std::os::unix::io::RawFd) -> Self {
+        Self::In(async_std::io::BufReader::new(
+            async_std::fs::File::from_raw_fd(fd),
+        ))
+    }
+
+    // Safety: fd must not be owned by any other File object
+    pub unsafe fn output(fd: std::os::unix::io::RawFd) -> Self {
+        Self::Out(async_std::fs::File::from_raw_fd(fd))
+    }
+
+    fn maybe_drop(file: crate::mutex::Mutex<Self>) {
+        if let Some(file) = crate::mutex::unwrap(file) {
+            if file.as_raw_fd() <= 2 {
+                let _ = file.into_raw_fd();
+            }
+        }
+    }
+}
+
+impl std::os::unix::io::AsRawFd for File {
+    fn as_raw_fd(&self) -> std::os::unix::io::RawFd {
+        match self {
+            Self::In(fh) => fh.get_ref().as_raw_fd(),
+            Self::Out(fh) => fh.as_raw_fd(),
+        }
+    }
+}
+
+impl std::os::unix::io::IntoRawFd for File {
+    fn into_raw_fd(self) -> std::os::unix::io::RawFd {
+        match self {
+            Self::In(fh) => fh.into_inner().into_raw_fd(),
+            Self::Out(fh) => fh.into_raw_fd(),
+        }
+    }
+}
+
+pub struct Child<'a> {
+    fut: std::pin::Pin<
+        Box<
+            dyn std::future::Future<Output = std::process::ExitStatus>
+                + Sync
+                + Send
+                + 'a,
+        >,
+    >,
+    wrapped_child: Option<Box<crate::runner::Child<'a>>>,
+}
+
+impl<'a> Child<'a> {
+    pub fn new_fut<F>(fut: F) -> Self
+    where
+        F: std::future::Future<Output = std::process::ExitStatus>
+            + Sync
+            + Send
+            + 'a,
+    {
+        Self {
+            fut: Box::pin(fut),
+            wrapped_child: None,
+        }
+    }
+
+    pub fn new_wrapped(child: crate::runner::Child<'a>) -> Self {
+        Self {
+            fut: Box::pin(async move { unreachable!() }),
+            wrapped_child: Some(Box::new(child)),
+        }
+    }
+
+    pub fn id(&self) -> Option<u32> {
+        self.wrapped_child.as_ref().and_then(|cmd| cmd.id())
+    }
+
+    pub fn status(
+        self,
+    ) -> std::pin::Pin<
+        Box<
+            dyn std::future::Future<
+                    Output = anyhow::Result<async_std::process::ExitStatus>,
+                > + Send
+                + Sync
+                + 'a,
+        >,
+    > {
+        Box::pin(async move {
+            if let Some(child) = self.wrapped_child {
+                child.status().await
+            } else {
+                Ok(self.fut.await)
+            }
+        })
+    }
+}
diff --git a/src/runner/builtins/mod.rs b/src/runner/builtins/mod.rs
new file mode 100644
index 0000000..dec4dc2
--- /dev/null
+++ b/src/runner/builtins/mod.rs
@@ -0,0 +1,314 @@
+use crate::runner::prelude::*;
+
+pub mod command;
+pub use command::{Child, Command, File, Io};
+
+type Builtin = &'static (dyn for<'a> Fn(
+    crate::parse::Exe,
+    &'a Env,
+    command::Cfg,
+) -> anyhow::Result<command::Child<'a>>
+              + Sync
+              + Send);
+
+#[allow(clippy::as_conversions)]
+static BUILTINS: once_cell::sync::Lazy<
+    std::collections::HashMap<&'static str, Builtin>,
+> = once_cell::sync::Lazy::new(|| {
+    let mut builtins = std::collections::HashMap::new();
+    builtins.insert("cd", &cd as Builtin);
+    builtins.insert("setenv", &setenv);
+    builtins.insert("unsetenv", &unsetenv);
+    builtins.insert("echo", &echo);
+    builtins.insert("read", &read);
+    builtins.insert("and", &and);
+    builtins.insert("or", &or);
+    builtins.insert("command", &command);
+    builtins.insert("builtin", &builtin);
+    builtins
+});
+
+macro_rules! bail {
+    ($cfg:expr, $exe:expr, $msg:expr $(,)?) => {
+        $cfg.io().write_stderr(
+            format!("{}: {}\n", $exe.exe().display(), $msg).as_bytes()
+        )
+        .await
+        .unwrap();
+        return std::process::ExitStatus::from_raw(1 << 8);
+    };
+    ($cfg:expr, $exe:expr, $msg:expr, $($arg:tt)*) => {
+        $cfg.io().write_stderr(
+            format!("{}: ", $exe.exe().display()).as_bytes()
+        )
+        .await
+        .unwrap();
+        $cfg.io().write_stderr(format!($msg, $($arg)*).as_bytes())
+            .await
+            .unwrap();
+        $cfg.io().write_stderr(b"\n").await.unwrap();
+        return std::process::ExitStatus::from_raw(1 << 8);
+    };
+}
+
+// clippy can't tell that the type is necessary
+#[allow(clippy::unnecessary_wraps)]
+fn cd(
+    exe: crate::parse::Exe,
+    env: &Env,
+    cfg: command::Cfg,
+) -> anyhow::Result<command::Child> {
+    async fn async_cd(
+        exe: crate::parse::Exe,
+        _env: &Env,
+        cfg: command::Cfg,
+    ) -> std::process::ExitStatus {
+        let dir = exe.args().get(0).map_or("", String::as_str);
+        let dir = if dir.is_empty() {
+            if let Some(dir) = home(None) {
+                dir
+            } else {
+                bail!(cfg, exe, "couldn't find current user");
+            }
+        } else if dir.starts_with('~') {
+            let path: std::path::PathBuf = dir.into();
+            if let std::path::Component::Normal(prefix) =
+                path.components().next().unwrap()
+            {
+                let prefix_bytes = prefix.as_bytes();
+                let name = if prefix_bytes == b"~" {
+                    None
+                } else {
+                    Some(std::ffi::OsStr::from_bytes(&prefix_bytes[1..]))
+                };
+                if let Some(home) = home(name) {
+                    home.join(path.strip_prefix(prefix).unwrap())
+                } else {
+                    bail!(
+                        cfg,
+                        exe,
+                        "no such user: {}",
+                        name.map(std::ffi::OsStr::to_string_lossy)
+                            .as_ref()
+                            .unwrap_or(&std::borrow::Cow::Borrowed(
+                                "(deleted)"
+                            ))
+                    );
+                }
+            } else {
+                unreachable!()
+            }
+        } else {
+            dir.into()
+        };
+        if let Err(e) = std::env::set_current_dir(&dir) {
+            bail!(
+                cfg,
+                exe,
+                "{}: {}",
+                crate::format::io_error(&e),
+                dir.display()
+            );
+        }
+        async_std::process::ExitStatus::from_raw(0)
+    }
+
+    Ok(command::Child::new_fut(async move {
+        async_cd(exe, env, cfg).await
+    }))
+}
+
+#[allow(clippy::unnecessary_wraps)]
+fn setenv(
+    exe: crate::parse::Exe,
+    env: &Env,
+    cfg: command::Cfg,
+) -> anyhow::Result<command::Child> {
+    async fn async_setenv(
+        exe: crate::parse::Exe,
+        _env: &Env,
+        cfg: command::Cfg,
+    ) -> std::process::ExitStatus {
+        let k = if let Some(k) = exe.args().get(0).map(String::as_str) {
+            k
+        } else {
+            bail!(cfg, exe, "usage: setenv key value");
+        };
+        let v = if let Some(v) = exe.args().get(1).map(String::as_str) {
+            v
+        } else {
+            bail!(cfg, exe, "usage: setenv key value");
+        };
+
+        std::env::set_var(k, v);
+        async_std::process::ExitStatus::from_raw(0)
+    }
+
+    Ok(command::Child::new_fut(async move {
+        async_setenv(exe, env, cfg).await
+    }))
+}
+
+#[allow(clippy::unnecessary_wraps)]
+fn unsetenv(
+    exe: crate::parse::Exe,
+    env: &Env,
+    cfg: command::Cfg,
+) -> anyhow::Result<command::Child> {
+    async fn async_unsetenv(
+        exe: crate::parse::Exe,
+        _env: &Env,
+        cfg: command::Cfg,
+    ) -> std::process::ExitStatus {
+        let k = if let Some(k) = exe.args().get(0).map(String::as_str) {
+            k
+        } else {
+            bail!(cfg, exe, "usage: unsetenv key");
+        };
+
+        std::env::remove_var(k);
+        async_std::process::ExitStatus::from_raw(0)
+    }
+
+    Ok(command::Child::new_fut(async move {
+        async_unsetenv(exe, env, cfg).await
+    }))
+}
+
+// clippy can't tell that the type is necessary
+#[allow(clippy::unnecessary_wraps)]
+// mostly just for testing and ensuring that builtins work, i'll likely remove
+// this later, since the binary seems totally fine
+fn echo(
+    exe: crate::parse::Exe,
+    env: &Env,
+    cfg: command::Cfg,
+) -> anyhow::Result<command::Child> {
+    async fn async_echo(
+        exe: crate::parse::Exe,
+        _env: &Env,
+        cfg: command::Cfg,
+    ) -> std::process::ExitStatus {
+        macro_rules! write_stdout {
+            ($bytes:expr) => {
+                if let Err(e) = cfg.io().write_stdout($bytes).await {
+                    cfg.io()
+                        .write_stderr(format!("echo: {}", e).as_bytes())
+                        .await
+                        .unwrap();
+                    return async_std::process::ExitStatus::from_raw(1 << 8);
+                }
+            };
+        }
+        let count = exe.args().len();
+        for (i, arg) in exe.args().iter().enumerate() {
+            write_stdout!(arg.as_bytes());
+            if i == count - 1 {
+                write_stdout!(b"\n");
+            } else {
+                write_stdout!(b" ");
+            }
+        }
+
+        async_std::process::ExitStatus::from_raw(0)
+    }
+
+    Ok(command::Child::new_fut(async move {
+        async_echo(exe, env, cfg).await
+    }))
+}
+
+#[allow(clippy::unnecessary_wraps)]
+fn read(
+    exe: crate::parse::Exe,
+    env: &Env,
+    cfg: command::Cfg,
+) -> anyhow::Result<command::Child> {
+    async fn async_read(
+        exe: crate::parse::Exe,
+        _env: &Env,
+        cfg: command::Cfg,
+    ) -> std::process::ExitStatus {
+        let var = if let Some(var) = exe.args().get(0).map(String::as_str) {
+            var
+        } else {
+            bail!(cfg, exe, "usage: read var");
+        };
+
+        let val = match cfg.io().read_line_stdin().await {
+            Ok(line) => line,
+            Err(e) => {
+                bail!(cfg, exe, e);
+            }
+        };
+
+        std::env::set_var(var, val);
+        async_std::process::ExitStatus::from_raw(0)
+    }
+
+    Ok(command::Child::new_fut(async move {
+        async_read(exe, env, cfg).await
+    }))
+}
+
+fn and(
+    mut exe: crate::parse::Exe,
+    env: &Env,
+    cfg: command::Cfg,
+) -> anyhow::Result<command::Child> {
+    exe.shift();
+    if env.latest_status().success() {
+        let mut cmd = crate::runner::Command::new(exe);
+        cfg.setup_command(&mut cmd);
+        Ok(command::Child::new_wrapped(cmd.spawn(env)?))
+    } else {
+        let status = *env.latest_status();
+        Ok(command::Child::new_fut(async move { status }))
+    }
+}
+
+fn or(
+    mut exe: crate::parse::Exe,
+    env: &Env,
+    cfg: command::Cfg,
+) -> anyhow::Result<command::Child> {
+    exe.shift();
+    if env.latest_status().success() {
+        let status = *env.latest_status();
+        Ok(command::Child::new_fut(async move { status }))
+    } else {
+        let mut cmd = crate::runner::Command::new(exe);
+        cfg.setup_command(&mut cmd);
+        Ok(command::Child::new_wrapped(cmd.spawn(env)?))
+    }
+}
+
+fn command(
+    mut exe: crate::parse::Exe,
+    env: &Env,
+    cfg: command::Cfg,
+) -> anyhow::Result<command::Child> {
+    exe.shift();
+    let mut cmd = crate::runner::Command::new_binary(exe);
+    cfg.setup_command(&mut cmd);
+    Ok(command::Child::new_wrapped(cmd.spawn(env)?))
+}
+
+fn builtin(
+    mut exe: crate::parse::Exe,
+    env: &Env,
+    cfg: command::Cfg,
+) -> anyhow::Result<command::Child> {
+    exe.shift();
+    let mut cmd = crate::runner::Command::new_builtin(exe);
+    cfg.setup_command(&mut cmd);
+    Ok(command::Child::new_wrapped(cmd.spawn(env)?))
+}
+
+fn home(user: Option<&std::ffi::OsStr>) -> Option<std::path::PathBuf> {
+    let user = user.map_or_else(
+        || users::get_user_by_uid(users::get_current_uid()),
+        users::get_user_by_name,
+    );
+    user.map(|user| user.home_dir().to_path_buf())
+}
diff --git a/src/runner/command.rs b/src/runner/command.rs
new file mode 100644
index 0000000..34b770e
--- /dev/null
+++ b/src/runner/command.rs
@@ -0,0 +1,219 @@
+use crate::runner::prelude::*;
+
+pub struct Command {
+    inner: Inner,
+    exe: std::path::PathBuf,
+    redirects: Vec<crate::parse::Redirect>,
+    pre_exec: Option<
+        Box<dyn FnMut() -> std::io::Result<()> + Send + Sync + 'static>,
+    >,
+}
+impl Command {
+    pub fn new(exe: crate::parse::Exe) -> Self {
+        let exe_path = exe.exe().to_path_buf();
+        let redirects = exe.redirects().to_vec();
+        Self {
+            inner: super::builtins::Command::new(exe).map_or_else(
+                |exe| Self::new_binary(exe).inner,
+                Inner::Builtin,
+            ),
+            exe: exe_path,
+            redirects,
+            pre_exec: None,
+        }
+    }
+
+    pub fn new_with_io(
+        exe: crate::parse::Exe,
+        io: super::builtins::Io,
+    ) -> Self {
+        let exe_path = exe.exe().to_path_buf();
+        let redirects = exe.redirects().to_vec();
+        Self {
+            inner: super::builtins::Command::new_with_io(exe, io)
+                .map_or_else(
+                    |exe| Self::new_binary(exe).inner,
+                    Inner::Builtin,
+                ),
+            exe: exe_path,
+            redirects,
+            pre_exec: None,
+        }
+    }
+
+    #[allow(clippy::needless_pass_by_value)]
+    pub fn new_binary(exe: crate::parse::Exe) -> Self {
+        let exe_path = exe.exe().to_path_buf();
+        let redirects = exe.redirects().to_vec();
+        let mut cmd = async_std::process::Command::new(exe.exe());
+        cmd.args(exe.args());
+        Self {
+            inner: Inner::Binary(cmd),
+            exe: exe_path,
+            redirects,
+            pre_exec: None,
+        }
+    }
+
+    pub fn new_builtin(exe: crate::parse::Exe) -> Self {
+        let exe_path = exe.exe().to_path_buf();
+        let redirects = exe.redirects().to_vec();
+        Self {
+            inner: super::builtins::Command::new(exe)
+                .map_or_else(|_| todo!(), Inner::Builtin),
+            exe: exe_path,
+            redirects,
+            pre_exec: None,
+        }
+    }
+
+    pub fn stdin(&mut self, fh: std::fs::File) {
+        match &mut self.inner {
+            Inner::Binary(cmd) => {
+                cmd.stdin(fh);
+            }
+            Inner::Builtin(cmd) => {
+                cmd.stdin(fh);
+            }
+        }
+    }
+
+    pub fn stdout(&mut self, fh: std::fs::File) {
+        match &mut self.inner {
+            Inner::Binary(cmd) => {
+                cmd.stdout(fh);
+            }
+            Inner::Builtin(cmd) => {
+                cmd.stdout(fh);
+            }
+        }
+    }
+
+    pub fn stderr(&mut self, fh: std::fs::File) {
+        match &mut self.inner {
+            Inner::Binary(cmd) => {
+                cmd.stderr(fh);
+            }
+            Inner::Builtin(cmd) => {
+                cmd.stderr(fh);
+            }
+        }
+    }
+
+    // Safety: see pre_exec in async_std::os::unix::process::CommandExt (this
+    // is just a wrapper)
+    pub unsafe fn pre_exec<F>(&mut self, f: F)
+    where
+        F: 'static + FnMut() -> std::io::Result<()> + Send + Sync,
+    {
+        self.pre_exec = Some(Box::new(f));
+    }
+
+    pub fn spawn(self, env: &Env) -> anyhow::Result<Child> {
+        let Self {
+            inner,
+            exe,
+            redirects,
+            pre_exec,
+        } = self;
+
+        #[allow(clippy::as_conversions)]
+        let pre_exec = pre_exec.map_or_else(
+            || {
+                let redirects = redirects.clone();
+                Box::new(move || {
+                    apply_redirects(&redirects)?;
+                    Ok(())
+                })
+                    as Box<dyn FnMut() -> std::io::Result<()> + Send + Sync>
+            },
+            |mut pre_exec| {
+                let redirects = redirects.clone();
+                Box::new(move || {
+                    apply_redirects(&redirects)?;
+                    pre_exec()?;
+                    Ok(())
+                })
+            },
+        );
+        match inner {
+            Inner::Binary(mut cmd) => {
+                // Safety: open, dup2, and close are async-signal-safe
+                // functions
+                unsafe { cmd.pre_exec(pre_exec) };
+                Ok(Child::Binary(cmd.spawn().map_err(|e| {
+                    anyhow::anyhow!(
+                        "{}: {}",
+                        crate::format::io_error(&e),
+                        exe.display()
+                    )
+                })?))
+            }
+            Inner::Builtin(mut cmd) => {
+                // Safety: open, dup2, and close are async-signal-safe
+                // functions
+                unsafe { cmd.pre_exec(pre_exec) };
+                cmd.apply_redirects(&redirects);
+                Ok(Child::Builtin(cmd.spawn(env)?))
+            }
+        }
+    }
+}
+
+pub enum Inner {
+    Binary(async_std::process::Command),
+    Builtin(super::builtins::Command),
+}
+
+pub enum Child<'a> {
+    Binary(async_std::process::Child),
+    Builtin(super::builtins::Child<'a>),
+}
+
+impl<'a> Child<'a> {
+    pub fn id(&self) -> Option<u32> {
+        match self {
+            Self::Binary(child) => Some(child.id()),
+            Self::Builtin(child) => child.id(),
+        }
+    }
+
+    pub fn status(
+        self,
+    ) -> std::pin::Pin<
+        Box<
+            dyn std::future::Future<
+                    Output = anyhow::Result<std::process::ExitStatus>,
+                > + Send
+                + Sync
+                + 'a,
+        >,
+    > {
+        Box::pin(async move {
+            match self {
+                Self::Binary(child) => Ok(child.status_no_drop().await?),
+                Self::Builtin(child) => Ok(child.status().await?),
+            }
+        })
+    }
+}
+
+fn apply_redirects(
+    redirects: &[crate::parse::Redirect],
+) -> std::io::Result<()> {
+    for redirect in redirects {
+        match &redirect.to {
+            crate::parse::RedirectTarget::Fd(fd) => {
+                nix::unistd::dup2(*fd, redirect.from)?;
+            }
+            crate::parse::RedirectTarget::File(path) => {
+                let fd = redirect.dir.open(path)?;
+                if fd != redirect.from {
+                    nix::unistd::dup2(fd, redirect.from)?;
+                    nix::unistd::close(fd)?;
+                }
+            }
+        }
+    }
+    Ok(())
+}
diff --git a/src/runner/mod.rs b/src/runner/mod.rs
new file mode 100644
index 0000000..3d4d025
--- /dev/null
+++ b/src/runner/mod.rs
@@ -0,0 +1,534 @@
+use crate::runner::prelude::*;
+
+mod builtins;
+mod command;
+pub use command::{Child, Command};
+mod prelude;
+
+const PID0: nix::unistd::Pid = nix::unistd::Pid::from_raw(0);
+
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
+pub enum Event {
+    Suspend(usize),
+    Exit(Env),
+}
+
+struct Stack {
+    frames: Vec<Frame>,
+}
+
+impl Stack {
+    fn new() -> Self {
+        Self { frames: vec![] }
+    }
+
+    fn push(&mut self, frame: Frame) {
+        self.frames.push(frame);
+    }
+
+    fn pop(&mut self) -> Frame {
+        self.frames.pop().unwrap()
+    }
+
+    fn top(&self) -> Option<&Frame> {
+        self.frames.last()
+    }
+
+    fn top_mut(&mut self) -> Option<&mut Frame> {
+        self.frames.last_mut()
+    }
+
+    fn current_pc(&self, pc: usize) -> bool {
+        match self.top() {
+            Some(Frame::If(_)) | None => false,
+            Some(Frame::While(_, start) | Frame::For(_, start, _)) => {
+                *start == pc
+            }
+        }
+    }
+
+    fn should_execute(&self) -> bool {
+        for frame in &self.frames {
+            if matches!(
+                frame,
+                Frame::If(false)
+                    | Frame::While(false, ..)
+                    | Frame::For(false, ..)
+            ) {
+                return false;
+            }
+        }
+        true
+    }
+}
+
+enum Frame {
+    If(bool),
+    While(bool, usize),
+    For(bool, usize, Vec<String>),
+}
+
+pub async fn run() -> anyhow::Result<i32> {
+    // Safety: we don't create File instances for or read/write data on fds
+    // 3 or 4 anywhere else
+    let shell_read = unsafe { async_std::fs::File::from_raw_fd(3) };
+    let shell_write = unsafe { async_std::fs::File::from_raw_fd(4) };
+    cloexec(3)?;
+    cloexec(4)?;
+
+    let (commands, mut env) = read_data(shell_read).await?;
+    run_commands(&commands, &mut env, &shell_write).await?;
+    let status = *env.latest_status();
+    write_event(&shell_write, Event::Exit(env)).await?;
+
+    if let Some(signal) = status.signal() {
+        nix::sys::signal::raise(signal.try_into().unwrap())?;
+    }
+    Ok(status.code().unwrap())
+}
+
+async fn run_commands(
+    commands: &str,
+    env: &mut Env,
+    shell_write: &async_std::fs::File,
+) -> anyhow::Result<()> {
+    let commands = crate::parse::ast::Commands::parse(commands)?;
+    let commands = commands.commands();
+    let mut pc = 0;
+    let mut stack = Stack::new();
+    while pc < commands.len() {
+        match &commands[pc] {
+            crate::parse::ast::Command::Pipeline(pipeline) => {
+                if stack.should_execute() {
+                    run_pipeline(pipeline.clone(), env, shell_write).await?;
+                }
+                pc += 1;
+            }
+            crate::parse::ast::Command::If(pipeline) => {
+                let should = stack.should_execute();
+                if !stack.current_pc(pc) {
+                    stack.push(Frame::If(false));
+                }
+                if should {
+                    run_pipeline(pipeline.clone(), env, shell_write).await?;
+                    if let Some(Frame::If(should)) = stack.top_mut() {
+                        *should = env.latest_status().success();
+                    } else {
+                        unreachable!();
+                    }
+                }
+                pc += 1;
+            }
+            crate::parse::ast::Command::While(pipeline) => {
+                let should = stack.should_execute();
+                if !stack.current_pc(pc) {
+                    stack.push(Frame::While(false, pc));
+                }
+                if should {
+                    run_pipeline(pipeline.clone(), env, shell_write).await?;
+                    if let Some(Frame::While(should, _)) = stack.top_mut() {
+                        *should = env.latest_status().success();
+                    } else {
+                        unreachable!();
+                    }
+                }
+                pc += 1;
+            }
+            crate::parse::ast::Command::For(var, list) => {
+                let should = stack.should_execute();
+                if !stack.current_pc(pc) {
+                    stack.push(Frame::For(
+                        false,
+                        pc,
+                        if stack.should_execute() {
+                            list.clone()
+                                .into_iter()
+                                .map(|w| w.eval(env))
+                                .collect()
+                        } else {
+                            vec![]
+                        },
+                    ));
+                }
+                if should {
+                    if let Some(Frame::For(should, _, list)) = stack.top_mut()
+                    {
+                        *should = !list.is_empty();
+                        if *should {
+                            let val = list.remove(0);
+                            env.set_var(var, &val);
+                        }
+                    } else {
+                        unreachable!();
+                    }
+                }
+                pc += 1;
+            }
+            crate::parse::ast::Command::End => match stack.top() {
+                Some(Frame::If(_)) => {
+                    stack.pop();
+                    pc += 1;
+                }
+                Some(
+                    Frame::While(should, start)
+                    | Frame::For(should, start, _),
+                ) => {
+                    if *should {
+                        pc = *start;
+                    } else {
+                        stack.pop();
+                        pc += 1;
+                    }
+                }
+                None => todo!(),
+            },
+        }
+    }
+    Ok(())
+}
+
+async fn run_pipeline(
+    pipeline: crate::parse::ast::Pipeline,
+    env: &mut Env,
+    shell_write: &async_std::fs::File,
+) -> anyhow::Result<()> {
+    // Safety: pipelines are run serially, so only one copy of these will ever
+    // exist at once. note that reusing a single copy of these at the top
+    // level would not be safe, because in the case of a command line like
+    // "echo foo; ls", we would pass the stdout fd to the ls process while it
+    // is still open here, and may still have data buffered.
+    let stdin = unsafe { async_std::fs::File::from_raw_fd(0) };
+    let stdout = unsafe { async_std::fs::File::from_raw_fd(1) };
+    let stderr = unsafe { async_std::fs::File::from_raw_fd(2) };
+    let mut io = builtins::Io::new();
+    io.set_stdin(stdin);
+    io.set_stdout(stdout);
+    io.set_stderr(stderr);
+
+    let (children, pg) = spawn_children(pipeline, env, &io)?;
+    let status = wait_children(children, pg, env, &io, shell_write).await;
+    set_foreground_pg(nix::unistd::getpid())?;
+    env.update(status)?;
+    Ok(())
+}
+
+async fn read_data(
+    mut fh: async_std::fs::File,
+) -> anyhow::Result<(String, Env)> {
+    let mut data = vec![];
+    fh.read_to_end(&mut data).await?;
+    let commands = bincode::deserialize(&data).unwrap();
+    let len: usize = bincode::serialized_size(&commands)
+        .unwrap()
+        .try_into()
+        .unwrap();
+    let env = Env::from_bytes(&data[len..]);
+    Ok((commands, env))
+}
+
+async fn write_event(
+    mut fh: &async_std::fs::File,
+    event: Event,
+) -> anyhow::Result<()> {
+    fh.write_all(&bincode::serialize(&event)?).await?;
+    fh.flush().await?;
+    Ok(())
+}
+
+fn spawn_children<'a>(
+    pipeline: crate::parse::ast::Pipeline,
+    env: &'a Env,
+    io: &builtins::Io,
+) -> anyhow::Result<(Vec<Child<'a>>, Option<nix::unistd::Pid>)> {
+    let pipeline = pipeline.eval(env);
+    let mut cmds: Vec<_> = pipeline
+        .into_exes()
+        .map(|exe| Command::new_with_io(exe, io.clone()))
+        .collect();
+    for i in 0..(cmds.len() - 1) {
+        let (r, w) = pipe()?;
+        cmds[i].stdout(w);
+        cmds[i + 1].stdin(r);
+    }
+
+    let mut children = vec![];
+    let mut pg_pid = None;
+    for mut cmd in cmds {
+        // Safety: setpgid is an async-signal-safe function
+        unsafe {
+            cmd.pre_exec(move || {
+                setpgid_child(pg_pid)?;
+                Ok(())
+            });
+        }
+        let child = cmd.spawn(env)?;
+        if let Some(id) = child.id() {
+            let child_pid = id_to_pid(id);
+            setpgid_parent(child_pid, pg_pid)?;
+            if pg_pid.is_none() {
+                pg_pid = Some(child_pid);
+                set_foreground_pg(child_pid)?;
+            }
+        }
+        children.push(child);
+    }
+    Ok((children, pg_pid))
+}
+
+async fn wait_children(
+    children: Vec<Child<'_>>,
+    pg: Option<nix::unistd::Pid>,
+    env: &Env,
+    io: &builtins::Io,
+    shell_write: &async_std::fs::File,
+) -> std::process::ExitStatus {
+    enum Res {
+        Child(nix::Result<nix::sys::wait::WaitStatus>),
+        Builtin(Option<(anyhow::Result<std::process::ExitStatus>, bool)>),
+    }
+
+    macro_rules! bail {
+        ($e:expr) => {
+            // if writing to stderr is not possible, we still want to exit
+            // normally with a failure exit code
+            #[allow(clippy::let_underscore_drop)]
+            let _ =
+                io.write_stderr(format!("nbsh: {}\n", $e).as_bytes()).await;
+            return std::process::ExitStatus::from_raw(1 << 8);
+        };
+    }
+
+    let mut final_status = None;
+
+    let count = children.len();
+    let (children, builtins): (Vec<_>, Vec<_>) = children
+        .into_iter()
+        .enumerate()
+        .partition(|(_, child)| child.id().is_some());
+    let mut children: std::collections::HashMap<_, _> = children
+        .into_iter()
+        .map(|(i, child)| {
+            (id_to_pid(child.id().unwrap()), (child, i == count - 1))
+        })
+        .collect();
+    let mut builtins: futures_util::stream::FuturesUnordered<_> =
+        builtins
+            .into_iter()
+            .map(|(i, child)| async move {
+                (child.status().await, i == count - 1)
+            })
+            .collect();
+
+    let (wait_w, wait_r) = async_std::channel::unbounded();
+    let new_wait = move || {
+        if let Some(pg) = pg {
+            let wait_w = wait_w.clone();
+            async_std::task::spawn(async move {
+                let res = blocking::unblock(move || {
+                    nix::sys::wait::waitpid(
+                        neg_pid(pg),
+                        Some(nix::sys::wait::WaitPidFlag::WUNTRACED),
+                    )
+                })
+                .await;
+                if wait_w.is_closed() {
+                    // we shouldn't be able to drop real process terminations
+                    assert!(res.is_err());
+                } else {
+                    wait_w.send(res).await.unwrap();
+                }
+            });
+        }
+    };
+
+    new_wait();
+    loop {
+        if children.is_empty() && builtins.is_empty() {
+            break;
+        }
+
+        let child = async { Res::Child(wait_r.recv().await.unwrap()) };
+        let builtin = async {
+            Res::Builtin(if builtins.is_empty() {
+                std::future::pending().await
+            } else {
+                builtins.next().await
+            })
+        };
+        match child.race(builtin).await {
+            Res::Child(Ok(status)) => {
+                match status {
+                    // we can't call child.status() here to unify these branches
+                    // because our waitpid call already collected the status
+                    nix::sys::wait::WaitStatus::Exited(pid, code) => {
+                        let (_, last) = children.remove(&pid).unwrap();
+                        if last {
+                            final_status = Some(
+                                std::process::ExitStatus::from_raw(code << 8),
+                            );
+                        }
+                    }
+                    nix::sys::wait::WaitStatus::Signaled(pid, signal, _) => {
+                        let (_, last) = children.remove(&pid).unwrap();
+                        if signal == nix::sys::signal::Signal::SIGINT {
+                            if let Err(e) = nix::sys::signal::raise(
+                                nix::sys::signal::Signal::SIGINT,
+                            ) {
+                                bail!(e);
+                            }
+                        }
+                        // this conversion is safe because the Signal enum is
+                        // repr(i32)
+                        #[allow(clippy::as_conversions)]
+                        if last {
+                            final_status =
+                                Some(std::process::ExitStatus::from_raw(
+                                    signal as i32,
+                                ));
+                        }
+                    }
+                    nix::sys::wait::WaitStatus::Stopped(pid, signal) => {
+                        if signal == nix::sys::signal::Signal::SIGTSTP {
+                            if let Err(e) = write_event(
+                                shell_write,
+                                Event::Suspend(env.idx()),
+                            )
+                            .await
+                            {
+                                bail!(e);
+                            }
+                            if let Err(e) = nix::sys::signal::kill(
+                                pid,
+                                nix::sys::signal::Signal::SIGCONT,
+                            ) {
+                                bail!(e);
+                            }
+                        }
+                    }
+                    _ => {}
+                }
+                new_wait();
+            }
+            Res::Child(Err(e)) => {
+                bail!(e);
+            }
+            Res::Builtin(Some((Ok(status), last))) => {
+                // this conversion is safe because the Signal enum is
+                // repr(i32)
+                #[allow(clippy::as_conversions)]
+                if status.signal()
+                    == Some(nix::sys::signal::Signal::SIGINT as i32)
+                {
+                    if let Err(e) = nix::sys::signal::raise(
+                        nix::sys::signal::Signal::SIGINT,
+                    ) {
+                        bail!(e);
+                    }
+                }
+                if last {
+                    final_status = Some(status);
+                }
+            }
+            Res::Builtin(Some((Err(e), _))) => {
+                bail!(e);
+            }
+            Res::Builtin(None) => {}
+        }
+    }
+
+    final_status.unwrap()
+}
+
+fn pipe() -> anyhow::Result<(std::fs::File, std::fs::File)> {
+    let (r, w) = nix::unistd::pipe2(nix::fcntl::OFlag::O_CLOEXEC)?;
+    // Safety: these file descriptors were just returned by pipe2 above, and
+    // are only available in this function, so nothing else can be accessing
+    // them
+    Ok((unsafe { std::fs::File::from_raw_fd(r) }, unsafe {
+        std::fs::File::from_raw_fd(w)
+    }))
+}
+
+fn set_foreground_pg(pg: nix::unistd::Pid) -> anyhow::Result<()> {
+    let pty = nix::fcntl::open(
+        "/dev/tty",
+        nix::fcntl::OFlag::empty(),
+        nix::sys::stat::Mode::empty(),
+    )?;
+
+    // if a background process calls tcsetpgrp, the kernel will send it
+    // SIGTTOU which suspends it. if that background process is the session
+    // leader and doesn't have SIGTTOU blocked, the kernel will instead just
+    // return ENOTTY from the tcsetpgrp call rather than sending a signal to
+    // avoid deadlocking the process. therefore, we need to ensure that
+    // SIGTTOU is blocked here.
+
+    // Safety: setting a signal handler to SigIgn is always safe
+    unsafe {
+        nix::sys::signal::signal(
+            nix::sys::signal::Signal::SIGTTOU,
+            nix::sys::signal::SigHandler::SigIgn,
+        )?;
+    }
+    let res = nix::unistd::tcsetpgrp(pty, pg);
+    // Safety: setting a signal handler to SigDfl is always safe
+    unsafe {
+        nix::sys::signal::signal(
+            nix::sys::signal::Signal::SIGTTOU,
+            nix::sys::signal::SigHandler::SigDfl,
+        )?;
+    }
+    res?;
+
+    nix::unistd::close(pty)?;
+
+    nix::sys::signal::kill(neg_pid(pg), nix::sys::signal::Signal::SIGCONT)
+        .or_else(|e| {
+            // the process group has already exited
+            if e == nix::errno::Errno::ESRCH {
+                Ok(())
+            } else {
+                Err(e)
+            }
+        })?;
+
+    Ok(())
+}
+
+fn setpgid_child(pg: Option<nix::unistd::Pid>) -> std::io::Result<()> {
+    nix::unistd::setpgid(PID0, pg.unwrap_or(PID0))?;
+    Ok(())
+}
+
+fn setpgid_parent(
+    pid: nix::unistd::Pid,
+    pg: Option<nix::unistd::Pid>,
+) -> anyhow::Result<()> {
+    nix::unistd::setpgid(pid, pg.unwrap_or(PID0)).or_else(|e| {
+        // EACCES means that the child already called exec, but if it did,
+        // then it also must have already called setpgid itself, so we don't
+        // care. ESRCH means that the process already exited, which is similar
+        if e == nix::errno::Errno::EACCES || e == nix::errno::Errno::ESRCH {
+            Ok(())
+        } else {
+            Err(e)
+        }
+    })?;
+    Ok(())
+}
+
+fn cloexec(fd: std::os::unix::io::RawFd) -> anyhow::Result<()> {
+    nix::fcntl::fcntl(
+        fd,
+        nix::fcntl::FcntlArg::F_SETFD(nix::fcntl::FdFlag::FD_CLOEXEC),
+    )?;
+    Ok(())
+}
+
+fn id_to_pid(id: u32) -> nix::unistd::Pid {
+    nix::unistd::Pid::from_raw(id.try_into().unwrap())
+}
+
+fn neg_pid(pid: nix::unistd::Pid) -> nix::unistd::Pid {
+    nix::unistd::Pid::from_raw(-pid.as_raw())
+}
diff --git a/src/runner/prelude.rs b/src/runner/prelude.rs
new file mode 100644
index 0000000..53b67fc
--- /dev/null
+++ b/src/runner/prelude.rs
@@ -0,0 +1 @@
+pub use crate::prelude::*;