diff --git a/.gitignore b/.gitignore
index 9976747ca57..eb72ae61208 100644
--- a/.gitignore
+++ b/.gitignore
@@ -54,3 +54,4 @@ GRTAGS
 GSYMS
 GTAGS
 .mypy_cache/
+/.pants
diff --git a/build-support/bin/rust/bootstrap_code.sh b/build-support/bin/rust/bootstrap_code.sh
index 4f6e9f2df6a..fc5eb4b6666 100644
--- a/build-support/bin/rust/bootstrap_code.sh
+++ b/build-support/bin/rust/bootstrap_code.sh
@@ -35,16 +35,19 @@ readonly NATIVE_ENGINE_BINARY_PYO3="native_engine_pyo3.so"
 readonly NATIVE_ENGINE_RESOURCE="${REPO_ROOT}/src/python/pants/engine/internals/${NATIVE_ENGINE_BINARY}"
 readonly NATIVE_ENGINE_RESOURCE_PYO3="${REPO_ROOT}/src/python/pants/engine/internals/${NATIVE_ENGINE_BINARY_PYO3}"
 readonly NATIVE_ENGINE_RESOURCE_METADATA="${NATIVE_ENGINE_RESOURCE}.metadata"
+readonly NATIVE_CLIENT_PATH="${REPO_ROOT}/.pants"
+readonly NATIVE_CLIENT_TARGET="${NATIVE_ROOT}/target/${MODE}/pants"
 
 function _build_native_code() {
-  banner "Building native engine..."
+  banner "Building native code..."
   # NB: See Cargo.toml with regard to the `extension-module` features.
   "${REPO_ROOT}/cargo" build \
     --features=extension-module \
     --features=engine_pyo3/extension-module \
     ${MODE_FLAG} \
     -p engine \
-    -p engine_pyo3 || die
+    -p engine_pyo3 \
+    -p client || die
 }
 
 function bootstrap_native_code() {
@@ -66,7 +69,11 @@ function bootstrap_native_code() {
   if [[ -f "${NATIVE_ENGINE_RESOURCE_METADATA}" ]]; then
     engine_version_in_metadata="$(sed -n 's/^engine_version: //p' "${NATIVE_ENGINE_RESOURCE_METADATA}")"
   fi
-  if [[ ! -f "${NATIVE_ENGINE_RESOURCE}" || ! -f "${NATIVE_ENGINE_RESOURCE_PYO3}" || "${engine_version_calculated}" != "${engine_version_in_metadata}" ]]; then
+  if [[ ! -f "${NATIVE_ENGINE_RESOURCE}" || ! -f \
+    "${NATIVE_ENGINE_RESOURCE_PYO3}" || ! -f \
+    "${NATIVE_CLIENT_PATH}" || \
+    "${engine_version_calculated}" != "${engine_version_in_metadata}" ]]; then
+
     _build_native_code || die
 
     # If bootstrapping the native engine fails, don't attempt to run pants
@@ -80,15 +87,21 @@ function bootstrap_native_code() {
       die "Failed to build native engine, file missing at ${native_binary_pyo3}."
     fi
 
+    # If bootstrapping the native client fails, don't attempt to run pants afterwards.
+    if [[ ! -f "${NATIVE_CLIENT_TARGET}" ]]; then
+      die "Failed to build native client."
+    fi
+
     # Pick up Cargo.lock changes if any caused by the `cargo build`.
     engine_version_calculated="$(calculate_current_hash)"
 
     # Create the native engine resource.
     # NB: On Mac Silicon, for some reason, first removing the old native_engine.so is necessary to avoid the Pants
     #  process from being killed when recompiling.
-    rm -f "${NATIVE_ENGINE_RESOURCE}" "${NATIVE_ENGINE_RESOURCE_PYO3}"
+    rm -f "${NATIVE_ENGINE_RESOURCE}" "${NATIVE_ENGINE_RESOURCE_PYO3}" "${NATIVE_CLIENT_PATH}"
     cp "${native_binary}" "${NATIVE_ENGINE_RESOURCE}"
     cp "${native_binary_pyo3}" "${NATIVE_ENGINE_RESOURCE_PYO3}"
+    cp "${NATIVE_CLIENT_TARGET}" "${NATIVE_CLIENT_PATH}"
 
     # Create the accompanying metadata file.
     local -r metadata_file=$(mktemp -t pants.native_engine.metadata.XXXXXX)
diff --git a/pants b/pants
index a95cc549a72..7ac3911dac3 100755
--- a/pants
+++ b/pants
@@ -38,17 +38,31 @@ source "${HERE}/build-support/pants_venv"
 source "${HERE}/build-support/bin/rust/bootstrap_code.sh"
 
 function exec_pants_bare() {
-
-  PANTS_EXE="${HERE}/src/python/pants/bin/pants_loader.py"
+  PANTS_NATIVE_EXE="${HERE}/.pants"
+  PANTS_PY_EXE="${HERE}/src/python/pants/bin/pants_loader.py"
   PANTS_SRCPATH="${HERE}/src/python"
 
   # Redirect activation and native bootstrap to ensure that they don't interfere with stdout.
   activate_pants_venv 1>&2
   bootstrap_native_code 1>&2
 
+  if [ -n "${USE_NATIVE_PANTS}" ]; then
+    set +e
+    "${PANTS_NATIVE_EXE}" "$@"
+    result=$?
+    # N.B.: The native pants client currently relies on pantsd being up. If it's not, it will fail
+    # with exit code 75 (EX_TEMPFAIL in /usr/include/sysexits.h) and we should fall through to the
+    # python pants client which knows how to start up pantsd. This failure takes O(1ms); so has no
+    # appreciable impact on --no-pantsd runs.
+    if ((result != 75)); then
+      exit ${result}
+    fi
+    set -e
+  fi
+
   # shellcheck disable=SC2086
   PYTHONPATH="${PANTS_SRCPATH}:${PYTHONPATH}" RUNNING_PANTS_FROM_SOURCES=1 \
-    exec ${PANTS_PREPEND_ARGS:-} "$(venv_dir)/bin/python" "${PANTS_EXE}" "$@"
+    exec ${PANTS_PREPEND_ARGS:-} "$(venv_dir)/bin/python" "${PANTS_PY_EXE}" "$@"
 }
 
 exec_pants_bare "$@"
diff --git a/src/rust/engine/Cargo.lock b/src/rust/engine/Cargo.lock
index 2f7ec93d6e4..0d716b5c5a6 100644
--- a/src/rust/engine/Cargo.lock
+++ b/src/rust/engine/Cargo.lock
@@ -320,6 +320,28 @@ dependencies = [
  "vec_map",
 ]
 
+[[package]]
+name = "client"
+version = "0.0.1"
+dependencies = [
+ "env_logger",
+ "futures",
+ "libc",
+ "log 0.4.11",
+ "nailgun",
+ "nix",
+ "peg",
+ "sha2",
+ "shellexpand",
+ "strum",
+ "strum_macros",
+ "sysinfo",
+ "tempdir",
+ "tokio",
+ "toml",
+ "uname",
+]
+
 [[package]]
 name = "cloudabi"
 version = "0.0.3"
@@ -607,6 +629,12 @@ dependencies = [
  "winapi 0.3.9",
 ]
 
+[[package]]
+name = "doc-comment"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fea41bba32d969b513997752735605054bc0dfa92b4c56bf1189f2e174be7a10"
+
 [[package]]
 name = "double-checked-cell-async"
 version = "2.0.2"
@@ -1468,9 +1496,9 @@ dependencies = [
 
 [[package]]
 name = "libc"
-version = "0.2.86"
+version = "0.2.95"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b7282d924be3275cec7f6756ff4121987bc6481325397dde6ba3e7802b1a8b1c"
+checksum = "789da6d93f1b866ffe175afc5322a4d76c038605a1c3319bb57b06967ca98a36"
 
 [[package]]
 name = "lmdb"
@@ -1754,6 +1782,18 @@ dependencies = [
  "winapi 0.3.9",
 ]
 
+[[package]]
+name = "nix"
+version = "0.20.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fa9b4819da1bc61c0ea48b63b7bc8604064dd43013e7cc325df098d49cd7c18a"
+dependencies = [
+ "bitflags",
+ "cc",
+ "cfg-if 1.0.0",
+ "libc",
+]
+
 [[package]]
 name = "nom"
 version = "6.0.1"
@@ -1994,6 +2034,33 @@ dependencies = [
  "proc-macro-hack",
 ]
 
+[[package]]
+name = "peg"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "07c0b841ea54f523f7aa556956fbd293bcbe06f2e67d2eb732b7278aaf1d166a"
+dependencies = [
+ "peg-macros",
+ "peg-runtime",
+]
+
+[[package]]
+name = "peg-macros"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b5aa52829b8decbef693af90202711348ab001456803ba2a98eb4ec8fb70844c"
+dependencies = [
+ "peg-runtime",
+ "proc-macro2 1.0.24",
+ "quote 1.0.8",
+]
+
+[[package]]
+name = "peg-runtime"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c719dcf55f09a3a7e764c6649ab594c18a177e3599c467983cdf644bfc0a4088"
+
 [[package]]
 name = "percent-encoding"
 version = "2.1.0"
@@ -2899,6 +2966,15 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4fb975283e6af8d3d691ddcefdca3a4dffe369167746d22fd993205e1e0c0de0"
 
+[[package]]
+name = "shellexpand"
+version = "2.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "83bdb7831b2d85ddf4a7b148aa19d0587eddbe8671a436b7bd1182eaad0f2829"
+dependencies = [
+ "dirs-next",
+]
+
 [[package]]
 name = "shlex"
 version = "0.1.1"
@@ -3110,6 +3186,22 @@ dependencies = [
  "unicode-xid 0.2.1",
 ]
 
+[[package]]
+name = "sysinfo"
+version = "0.17.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e9d811a7fd68bad355ff84f7d8ad58b33415bccdfce499f98a9ce30471888e05"
+dependencies = [
+ "cfg-if 1.0.0",
+ "core-foundation-sys",
+ "doc-comment",
+ "libc",
+ "ntapi",
+ "once_cell",
+ "rayon",
+ "winapi 0.3.9",
+]
+
 [[package]]
 name = "tap"
 version = "1.0.0"
@@ -3128,6 +3220,16 @@ dependencies = [
  "workunit_store",
 ]
 
+[[package]]
+name = "tempdir"
+version = "0.3.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "15f2b5fb00ccdf689e0149d1b1b3c03fead81c2b37735d812fa8bddbbf41b6d8"
+dependencies = [
+ "rand 0.4.6",
+ "remove_dir_all",
+]
+
 [[package]]
 name = "tempfile"
 version = "3.2.0"
diff --git a/src/rust/engine/Cargo.toml b/src/rust/engine/Cargo.toml
index 7620ec1b70e..577b18b228e 100644
--- a/src/rust/engine/Cargo.toml
+++ b/src/rust/engine/Cargo.toml
@@ -22,6 +22,7 @@ members = [
   "async_latch",
   "async_semaphore",
   "async_value",
+  "client",
   "concrete_time",
   "engine_pyo3",
   "fs",
@@ -60,6 +61,7 @@ default-members = [
   ".",
   "async_latch",
   "async_semaphore",
+  "client",
   "async_value",
   "concrete_time",
   "engine_pyo3",
diff --git a/src/rust/engine/client/Cargo.toml b/src/rust/engine/client/Cargo.toml
new file mode 100644
index 00000000000..c59b44f052c
--- /dev/null
+++ b/src/rust/engine/client/Cargo.toml
@@ -0,0 +1,30 @@
+[package]
+version = "0.0.1"
+edition = "2018"
+name = "client"
+authors = [ "Pants Build <pantsbuild@gmail.com>" ]
+publish = false
+
+[[bin]]
+name = "pants"
+path = "src/main.rs"
+
+[dependencies]
+env_logger = "0.5.4"
+futures = "0.3"
+libc = "0.2"
+log = "0.4"
+nailgun = { path = "../nailgun" }
+nix = "0.20"
+peg = "0.7"
+sha2 = "0.9"
+shellexpand = "2.1"
+strum = "0.20"
+strum_macros = "0.20"
+sysinfo = "0.17.1"
+tokio = { version = "1.4", features = ["rt-multi-thread", "macros", "net", "io-std", "io-util"] }
+toml = "0.5"
+uname = "0.1"
+
+[dev-dependencies]
+tempdir = "0.3"
\ No newline at end of file
diff --git a/src/rust/engine/client/src/build_root.rs b/src/rust/engine/client/src/build_root.rs
new file mode 100644
index 00000000000..d279cf4dd8d
--- /dev/null
+++ b/src/rust/engine/client/src/build_root.rs
@@ -0,0 +1,64 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use std::env;
+use std::ops::Deref;
+use std::path::{Path, PathBuf};
+
+use log::debug;
+
+#[derive(Debug)]
+pub struct BuildRoot(PathBuf);
+
+impl BuildRoot {
+  const SENTINEL_FILES: &'static [&'static str] = &["pants", "BUILDROOT", "BUILD_ROOT"];
+
+  pub fn find() -> Result<BuildRoot, String> {
+    let cwd = env::current_dir().map_err(|e| format!("Failed to determine $CWD: {}", e))?;
+    Self::find_from(&cwd)
+  }
+
+  pub(crate) fn find_from(start: &Path) -> Result<BuildRoot, String> {
+    let mut build_root = start;
+    loop {
+      for sentinel in Self::SENTINEL_FILES {
+        let sentinel_path = build_root.join(sentinel);
+        if !sentinel_path.exists() {
+          continue;
+        }
+        let sentinel_path_metadata = sentinel_path.metadata().map_err(|e| {
+          format!(
+            "\
+            Failed to read metadata for {path} to determine if is a build root sentinel file: {err}\
+            ",
+            path = sentinel_path.display(),
+            err = e
+          )
+        })?;
+        if sentinel_path_metadata.is_file() {
+          let root = BuildRoot(build_root.to_path_buf());
+          debug!("Found {:?} starting search from {}.", root, start.display());
+          return Ok(root);
+        }
+      }
+
+      build_root = build_root.parent().ok_or(format!(
+        "\
+          No build root detected for the current directory of {cwd}. Pants detects the build root \
+          by looking for at least one file from {sentinel_files} in the cwd and its ancestors. If \
+          you have none of these files, you can create an empty file in your build root.\
+          ",
+        cwd = start.display(),
+        sentinel_files = Self::SENTINEL_FILES.join(", ")
+      ))?;
+    }
+  }
+}
+
+impl Deref for BuildRoot {
+  type Target = PathBuf;
+
+  fn deref(&self) -> &PathBuf {
+    &self.0
+  }
+}
diff --git a/src/rust/engine/client/src/build_root_tests.rs b/src/rust/engine/client/src/build_root_tests.rs
new file mode 100644
index 00000000000..30cc9f7c2cc
--- /dev/null
+++ b/src/rust/engine/client/src/build_root_tests.rs
@@ -0,0 +1,57 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use std::fs;
+use std::path::PathBuf;
+
+use tempdir::TempDir;
+
+use crate::build_root::BuildRoot;
+use std::ops::Deref;
+
+#[test]
+fn test_find_cwd() {
+  let buildroot = TempDir::new("buildroot").unwrap();
+  let buildroot_path = buildroot.path().to_path_buf();
+  let mut sentinel: Option<PathBuf> = None;
+
+  let mut assert_sentinel = |name| {
+    if let Some(prior_sentinel) = sentinel.take() {
+      fs::remove_file(prior_sentinel).unwrap();
+    }
+    assert!(BuildRoot::find_from(&buildroot_path).is_err());
+
+    let file = buildroot.path().join(name);
+    fs::write(&file, &[]).unwrap();
+    sentinel = Some(file);
+    assert_eq!(
+      &buildroot_path,
+      BuildRoot::find_from(&buildroot_path).unwrap().deref()
+    );
+  };
+
+  assert_sentinel("BUILDROOT");
+  assert_sentinel("BUILD_ROOT");
+  assert_sentinel("pants");
+}
+
+#[test]
+fn test_find_subdir() {
+  let buildroot = TempDir::new("buildroot").unwrap();
+  let buildroot_path = buildroot.path().to_path_buf();
+  let subdir = buildroot_path.join("foo").join("bar");
+
+  assert!(BuildRoot::find_from(&buildroot_path).is_err());
+  assert!(BuildRoot::find_from(&subdir).is_err());
+
+  let sentinel = &buildroot.path().join("pants");
+  fs::write(&sentinel, &[]).unwrap();
+  assert_eq!(
+    &buildroot_path,
+    BuildRoot::find_from(&buildroot_path).unwrap().deref()
+  );
+  assert_eq!(
+    &buildroot_path,
+    BuildRoot::find_from(&subdir).unwrap().deref()
+  );
+}
diff --git a/src/rust/engine/client/src/client.rs b/src/rust/engine/client/src/client.rs
new file mode 100644
index 00000000000..ea6601f904d
--- /dev/null
+++ b/src/rust/engine/client/src/client.rs
@@ -0,0 +1,108 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use std::os::unix::io::AsRawFd;
+use std::time::SystemTime;
+
+use log::debug;
+
+use nailgun::NailgunClientError;
+
+pub struct ConnectionSettings {
+  pub port: u16,
+  pub timeout_limit: f64,
+  pub dynamic_ui: bool,
+}
+
+impl ConnectionSettings {
+  pub fn new(port: u16) -> ConnectionSettings {
+    ConnectionSettings {
+      port,
+      timeout_limit: 60.0,
+      dynamic_ui: true,
+    }
+  }
+}
+
+pub async fn execute_command(
+  start: SystemTime,
+  connection_settings: ConnectionSettings,
+  mut env: Vec<(String, String)>,
+  argv: Vec<String>,
+) -> Result<i32, String> {
+  env.push((
+    "PANTSD_RUNTRACKER_CLIENT_START_TIME".to_owned(),
+    start
+      .duration_since(SystemTime::UNIX_EPOCH)
+      .map_err(|e| format!("Failed to determine current time: {err}", err = e))?
+      .as_secs_f64()
+      .to_string(),
+  ));
+
+  env.push((
+    "PANTSD_REQUEST_TIMEOUT_LIMIT".to_owned(),
+    connection_settings.timeout_limit.to_string(),
+  ));
+
+  let raw_io_fds = [
+    std::io::stdin().as_raw_fd(),
+    std::io::stdout().as_raw_fd(),
+    std::io::stderr().as_raw_fd(),
+  ];
+  let mut tty_settings = Vec::with_capacity(raw_io_fds.len());
+  for raw_fd in &raw_io_fds {
+    match nix::sys::termios::tcgetattr(*raw_fd) {
+      Ok(termios) => tty_settings.push((raw_fd, termios)),
+      Err(err) => debug!(
+        "Failed to save terminal attributes for file descriptor {fd}: {err}",
+        fd = raw_fd,
+        err = err
+      ),
+    }
+    if connection_settings.dynamic_ui {
+      if let Ok(path) = nix::unistd::ttyname(*raw_fd) {
+        env.push((
+          format!("NAILGUN_TTY_PATH_{fd}", fd = raw_fd),
+          path.display().to_string(),
+        ));
+      }
+    }
+  }
+
+  let command = argv
+    .get(0)
+    .ok_or_else(|| "Failed to determine current process argv0".to_owned())?
+    .clone();
+
+  let args = argv.iter().skip(1).cloned().collect();
+
+  let nailgun_result = nailgun::client_execute(connection_settings.port, command, args, env).await;
+  for (raw_fd, termios) in tty_settings {
+    if let Err(err) =
+      nix::sys::termios::tcsetattr(*raw_fd, nix::sys::termios::SetArg::TCSADRAIN, &termios)
+    {
+      debug!(
+        "Failed to restore terminal attributes for file descriptor {fd}: {err}",
+        fd = raw_fd,
+        err = err
+      );
+    }
+  }
+  nailgun_result.map_err(|error| match error {
+    NailgunClientError::PreConnect(err) => format!(
+      "Problem connecting to pantsd at {port}: {err}",
+      port = connection_settings.port,
+      err = err
+    ),
+    NailgunClientError::PostConnect(err) => format!(
+      "Problem communicating with pantsd at {port}: {err}",
+      port = connection_settings.port,
+      err = err
+    ),
+    NailgunClientError::BrokenPipe => format!(
+      "Broken pipe communicating with pantsd at {port}.",
+      port = connection_settings.port
+    ),
+    NailgunClientError::KeyboardInterrupt => "User interrupt.".to_owned(),
+  })
+}
diff --git a/src/rust/engine/client/src/client_tests.rs b/src/rust/engine/client/src/client_tests.rs
new file mode 100644
index 00000000000..22af6b8cdb6
--- /dev/null
+++ b/src/rust/engine/client/src/client_tests.rs
@@ -0,0 +1,22 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use crate::pantsd_testing::launch_pantsd;
+use crate::{execute_command, pantsd, ConnectionSettings};
+use std::time::SystemTime;
+
+#[tokio::test]
+async fn test_client() {
+  let (build_root, pants_subprocessdir) = launch_pantsd();
+
+  let port = pantsd::probe(&build_root, pants_subprocessdir.path()).unwrap();
+  let exit_code = execute_command(
+    SystemTime::now(),
+    ConnectionSettings::new(port),
+    std::env::vars().collect(),
+    ["pants", "-V"].iter().map(ToString::to_string).collect(),
+  )
+  .await
+  .unwrap();
+  assert_eq!(0, exit_code)
+}
diff --git a/src/rust/engine/client/src/lib.rs b/src/rust/engine/client/src/lib.rs
new file mode 100644
index 00000000000..88d2f70823e
--- /dev/null
+++ b/src/rust/engine/client/src/lib.rs
@@ -0,0 +1,55 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+#![deny(warnings)]
+// Enable all clippy lints except for many of the pedantic ones. It's a shame this needs to be copied and pasted across crates, but there doesn't appear to be a way to include inner attributes from a common source.
+#![deny(
+  clippy::all,
+  clippy::default_trait_access,
+  clippy::expl_impl_clone_on_copy,
+  clippy::if_not_else,
+  clippy::needless_continue,
+  clippy::unseparated_literal_suffix,
+  // TODO: Falsely triggers for async/await:
+  //   see https://github.com/rust-lang/rust-clippy/issues/5360
+  // clippy::used_underscore_binding
+)]
+// It is often more clear to show that nothing is being moved.
+#![allow(clippy::match_ref_pats)]
+// Subjective style.
+#![allow(
+  clippy::len_without_is_empty,
+  clippy::redundant_field_names,
+  clippy::too_many_arguments
+)]
+// Default isn't as big a deal as people seem to think it is.
+#![allow(clippy::new_without_default, clippy::new_ret_no_self)]
+// Arc<Mutex> can be more clear than needing to grok Orderings:
+#![allow(clippy::mutex_atomic)]
+
+mod build_root;
+#[cfg(test)]
+mod build_root_tests;
+
+mod client;
+#[cfg(test)]
+mod client_tests;
+
+pub mod options;
+
+pub mod pantsd;
+#[cfg(test)]
+mod pantsd_testing;
+#[cfg(test)]
+mod pantsd_tests;
+
+pub use crate::client::{execute_command, ConnectionSettings};
+
+pub fn render_choice(items: &[&str]) -> Option<String> {
+  match items {
+    [] => None,
+    [this] => Some(this.to_string()),
+    [this, that] => Some(format!("{} or {}", this, that)),
+    [these @ .., that] => Some(format!("{} or {}", these.join(", "), that)),
+  }
+}
diff --git a/src/rust/engine/client/src/main.rs b/src/rust/engine/client/src/main.rs
new file mode 100644
index 00000000000..83b41545f66
--- /dev/null
+++ b/src/rust/engine/client/src/main.rs
@@ -0,0 +1,156 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+#![deny(warnings)]
+// Enable all clippy lints except for many of the pedantic ones. It's a shame this needs to be copied and pasted across crates, but there doesn't appear to be a way to include inner attributes from a common source.
+#![deny(
+  clippy::all,
+  clippy::default_trait_access,
+  clippy::expl_impl_clone_on_copy,
+  clippy::if_not_else,
+  clippy::needless_continue,
+  clippy::unseparated_literal_suffix,
+  // TODO: Falsely triggers for async/await:
+  //   see https://github.com/rust-lang/rust-clippy/issues/5360
+  // clippy::used_underscore_binding
+)]
+// It is often more clear to show that nothing is being moved.
+#![allow(clippy::match_ref_pats)]
+// Subjective style.
+#![allow(
+  clippy::len_without_is_empty,
+  clippy::redundant_field_names,
+  clippy::too_many_arguments
+)]
+// Default isn't as big a deal as people seem to think it is.
+#![allow(clippy::new_without_default, clippy::new_ret_no_self)]
+// Arc<Mutex> can be more clear than needing to grok Orderings:
+#![allow(clippy::mutex_atomic)]
+
+use std::convert::AsRef;
+use std::env;
+use std::path::{Path, PathBuf};
+use std::str::FromStr;
+use std::time::SystemTime;
+
+use log::debug;
+use strum::VariantNames;
+use strum_macros::{AsRefStr, EnumString, EnumVariantNames};
+
+use client::option_id;
+use client::options::OptionParser;
+use client::pantsd;
+use client::render_choice;
+
+// TODO(John Sirois): Maybe consolidate with PythonLogLevel in src/rust/engine/logging/src/lib.rs.
+#[derive(AsRefStr, EnumString, EnumVariantNames)]
+#[strum(serialize_all = "snake_case")]
+enum PythonLogLevel {
+  Trace,
+  Debug,
+  Info,
+  Warn,
+  Error,
+}
+
+async fn execute(start: SystemTime) -> Result<i32, String> {
+  let options_parser = OptionParser::new()?;
+
+  let use_pantsd = options_parser.parse_bool(&option_id!("pantsd"), true)?;
+  if !use_pantsd.value {
+    return Err(format!(
+      "Pantsd has been turned off via {option_source:?}.",
+      option_source = use_pantsd.source
+    ));
+  }
+
+  let level_option = option_id!(-'l', "level");
+  let log_level_option_value =
+    options_parser.parse_string(&level_option, PythonLogLevel::Info.as_ref())?;
+  let level = PythonLogLevel::from_str(&log_level_option_value.value).map_err(|_| {
+    format!(
+      "Not a valid log level {level} from {option_source:?}. Should be one of {levels}.",
+      level = log_level_option_value.value,
+      option_source = log_level_option_value.source,
+      levels = render_choice(PythonLogLevel::VARIANTS)
+        .expect("We know there is at least one PythonLogLevel enum variant."),
+    )
+  })?;
+  env_logger::init_from_env(env_logger::Env::new().filter_or("__PANTS_LEVEL__", level.as_ref()));
+
+  let working_dir = env::current_dir()
+    .map_err(|e| format!("Could not detect current working directory: {err}", err = e))?;
+  let pantsd_settings = find_pantsd(&working_dir, &options_parser)?;
+  let env = env::vars().collect::<Vec<(_, _)>>();
+  let argv = env::args().collect::<Vec<_>>();
+  client::execute_command(start, pantsd_settings, env, argv).await
+}
+
+fn find_pantsd(
+  working_dir: &Path,
+  options_parser: &OptionParser,
+) -> Result<client::ConnectionSettings, String> {
+  let pants_subprocessdir = option_id!("pants", "subprocessdir");
+  let option_value = options_parser.parse_string(&pants_subprocessdir, ".pids")?;
+  let metadata_dir = {
+    let path = PathBuf::from(&option_value.value);
+    if path.is_absolute() {
+      path
+    } else {
+      match working_dir.join(&path) {
+        p if p.is_absolute() => p,
+        p => p.canonicalize().map_err(|e| {
+          format!(
+            "Failed to resolve relative pants subprocessdir specified via {:?} as {}: {}",
+            option_value,
+            path.display(),
+            e
+          )
+        })?,
+      }
+    }
+  };
+  debug!(
+    "\
+    Looking for pantsd metadata in {metadata_dir} as specified by {option} = {value} via \
+    {source:?}.\
+    ",
+    metadata_dir = metadata_dir.display(),
+    option = pants_subprocessdir,
+    value = option_value.value,
+    source = option_value.source
+  );
+  let port = pantsd::probe(&working_dir, &metadata_dir)?;
+  let mut pantsd_settings = client::ConnectionSettings::new(port);
+  pantsd_settings.timeout_limit = options_parser
+    .parse_float(
+      &option_id!("pantsd", "timeout", "when", "multiple", "invocations"),
+      pantsd_settings.timeout_limit,
+    )?
+    .value;
+  pantsd_settings.dynamic_ui = options_parser
+    .parse_bool(&option_id!("dynamic", "ui"), pantsd_settings.dynamic_ui)?
+    .value;
+  Ok(pantsd_settings)
+}
+
+// The value is taken from this C precedent:
+// ```
+// $ grep 75 /usr/include/sysexits.h
+// #define EX_TEMPFAIL	75	/* temp failure; user is invited to retry */
+// ```
+const EX_TEMPFAIL: i32 = 75;
+
+#[tokio::main]
+async fn main() {
+  let start = SystemTime::now();
+  match execute(start).await {
+    Err(err) => {
+      eprintln!("{}", err);
+      // We use this exit code to indicate an error running pants via the nailgun protocol to
+      // differentiate from a successful nailgun protocol session.
+      std::process::exit(EX_TEMPFAIL);
+    }
+    Ok(exit_code) => std::process::exit(exit_code),
+  }
+}
diff --git a/src/rust/engine/client/src/options/args.rs b/src/rust/engine/client/src/options/args.rs
new file mode 100644
index 00000000000..b456c7fb5d3
--- /dev/null
+++ b/src/rust/engine/client/src/options/args.rs
@@ -0,0 +1,125 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use std::env;
+
+use super::id::{NameTransform, OptionId, Scope};
+use super::parse::parse_bool;
+use super::OptionsSource;
+use crate::options::parse::parse_string_list;
+use crate::options::ListEdit;
+use std::collections::HashMap;
+
+pub(crate) struct Args {
+  pub(crate) args: Vec<String>,
+}
+
+#[derive(PartialEq)]
+enum Negate {
+  True,
+  False,
+}
+
+impl Args {
+  pub(crate) fn argv() -> Args {
+    Args {
+      args: env::args().collect::<Vec<_>>(),
+    }
+  }
+
+  fn arg_name(id: &OptionId, negate: Negate) -> String {
+    format!(
+      "--{}{}{}",
+      match negate {
+        Negate::False => "",
+        Negate::True => "no-",
+      },
+      match &id.0 {
+        Scope::Global => "".to_string(),
+        Scope::Scope(scope) => format!("{}-", scope.to_ascii_lowercase()),
+      },
+      id.name("-", NameTransform::ToLower)
+    )
+  }
+
+  fn arg_names(id: &OptionId, negate: Negate) -> HashMap<String, bool> {
+    let mut arg_names = HashMap::new();
+    if let Some(switch) = id.2 {
+      arg_names.insert(format!("-{}", switch), false);
+      if negate == Negate::True {
+        arg_names.insert(format!("--no-{}", switch), true);
+      }
+    }
+    arg_names.insert(Self::arg_name(id, Negate::False), false);
+    if negate == Negate::True {
+      arg_names.insert(Self::arg_name(id, Negate::True), true);
+    }
+    arg_names
+  }
+
+  fn find_flag(
+    &self,
+    flag_names: HashMap<String, bool>,
+  ) -> Result<Option<(String, String, bool)>, String> {
+    for arg in self.args.iter().rev() {
+      let mut components = arg.as_str().splitn(2, '=');
+      if let Some(name) = components.next() {
+        if let Some(negated) = flag_names.get(name) {
+          return Ok(Some((
+            name.to_owned(),
+            components.next().unwrap_or("").to_owned(),
+            *negated,
+          )));
+        }
+      }
+    }
+    Ok(None)
+  }
+}
+
+impl OptionsSource for Args {
+  fn display(&self, id: &OptionId) -> String {
+    Self::arg_name(id, Negate::False)
+  }
+
+  fn get_string(&self, id: &OptionId) -> Result<Option<String>, String> {
+    self
+      .find_flag(Self::arg_names(id, Negate::False))
+      .map(|value| value.map(|(_, v, _)| v))
+  }
+
+  fn get_bool(&self, id: &OptionId) -> Result<Option<bool>, String> {
+    let arg_names = Self::arg_names(id, Negate::True);
+    match self.find_flag(arg_names)? {
+      Some((_, s, negated)) if s.as_str() == "" => Ok(Some(!negated)),
+      Some((name, ref value, negated)) => parse_bool(value)
+        .map(|b| Some(b ^ negated))
+        .map_err(|e| e.render(name)),
+      None => Ok(None),
+    }
+  }
+
+  fn get_string_list(&self, id: &OptionId) -> Result<Option<Vec<ListEdit<String>>>, String> {
+    let arg_names = Self::arg_names(id, Negate::False);
+    let mut edits = vec![];
+    for arg in &self.args {
+      let mut components = arg.as_str().splitn(2, '=');
+      if let Some(name) = components.next() {
+        if arg_names.contains_key(name) {
+          let value = components.next().ok_or_else(|| {
+            format!(
+              "Expected string list option {name} to have a value.",
+              name = name
+            )
+          })?;
+          edits.extend(parse_string_list(value).map_err(|e| e.render(name))?)
+        }
+      }
+    }
+    if edits.is_empty() {
+      Ok(None)
+    } else {
+      Ok(Some(edits))
+    }
+  }
+}
diff --git a/src/rust/engine/client/src/options/args_tests.rs b/src/rust/engine/client/src/options/args_tests.rs
new file mode 100644
index 00000000000..e15f32f4fd3
--- /dev/null
+++ b/src/rust/engine/client/src/options/args_tests.rs
@@ -0,0 +1,150 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use crate::option_id;
+use crate::options::args::Args;
+use crate::options::{ListEdit, ListEditAction, OptionId, OptionsSource};
+
+fn args<I: IntoIterator<Item = &'static str>>(args: I) -> Args {
+  Args {
+    args: args.into_iter().map(str::to_owned).collect(),
+  }
+}
+
+#[test]
+fn test_display() {
+  let args = args([]);
+  assert_eq!("--global".to_owned(), args.display(&option_id!("global")));
+  assert_eq!(
+    "--scope-name".to_owned(),
+    args.display(&option_id!("scope", "name"))
+  );
+  assert_eq!(
+    "--scope-full-name".to_owned(),
+    args.display(&option_id!(-'f', "scope", "full", "name"))
+  );
+}
+
+#[test]
+fn test_string() {
+  let args = args([
+    "-u=swallow",
+    "--foo=bar",
+    "--baz-spam=eggs",
+    "--baz-spam=cheese",
+  ]);
+
+  let assert_string = |expected: &str, id: OptionId| {
+    assert_eq!(expected.to_owned(), args.get_string(&id).unwrap().unwrap())
+  };
+
+  assert_string("bar", option_id!("foo"));
+  assert_string("cheese", option_id!("baz", "spam"));
+  assert_string("swallow", option_id!(-'u', "unladen", "capacity"));
+
+  assert!(args.get_string(&option_id!("dne")).unwrap().is_none());
+}
+
+#[test]
+fn test_bool() {
+  let args = args([
+    "-c=swallow",
+    "--foo=false",
+    "-f",
+    "--no-bar",
+    "--baz=true",
+    "--baz=FALSE",
+    "--no-spam-eggs=False",
+    "--no-b=True",
+  ]);
+
+  let assert_bool =
+    |expected: bool, id: OptionId| assert_eq!(expected, args.get_bool(&id).unwrap().unwrap());
+
+  assert_bool(true, option_id!(-'f', "foo"));
+  assert_bool(false, option_id!("bar"));
+  assert_bool(false, option_id!(-'b', "baz"));
+  assert_bool(true, option_id!("spam", "eggs"));
+
+  assert!(args.get_bool(&option_id!("dne")).unwrap().is_none());
+  assert_eq!(
+    "Got 'swallow' for -c. Expected 'true' or 'false'.".to_owned(),
+    args
+      .get_bool(&option_id!(-'c', "unladen", "capacity"))
+      .unwrap_err()
+  );
+}
+
+#[test]
+fn test_float() {
+  let args = args([
+    "-j=4",
+    "--foo=42",
+    "--foo=3.14",
+    "--baz-spam=1.137",
+    "--bad=swallow",
+  ]);
+
+  let assert_float =
+    |expected: f64, id: OptionId| assert_eq!(expected, args.get_float(&id).unwrap().unwrap());
+
+  assert_float(4 as f64, option_id!(-'j', "jobs"));
+  assert_float(3.14, option_id!("foo"));
+  assert_float(1.137, option_id!("baz", "spam"));
+
+  assert!(args.get_float(&option_id!("dne")).unwrap().is_none());
+
+  assert_eq!(
+    "Problem parsing --bad value swallow as a float value: invalid float literal".to_owned(),
+    args.get_float(&option_id!("bad")).unwrap_err()
+  );
+}
+
+#[test]
+fn test_string_list() {
+  let args = args([
+    "--bad=['mis', 'matched')",
+    "--phases=initial",
+    "-p=['one']",
+    "--phases=+['two','three'],-['one']",
+  ]);
+
+  assert_eq!(
+    vec![
+      ListEdit {
+        action: ListEditAction::Add,
+        items: vec!["initial".to_owned()]
+      },
+      ListEdit {
+        action: ListEditAction::Replace,
+        items: vec!["one".to_owned()]
+      },
+      ListEdit {
+        action: ListEditAction::Add,
+        items: vec!["two".to_owned(), "three".to_owned()]
+      },
+      ListEdit {
+        action: ListEditAction::Remove,
+        items: vec!["one".to_owned()]
+      },
+    ],
+    args
+      .get_string_list(&option_id!(-'p', "phases"))
+      .unwrap()
+      .unwrap()
+  );
+
+  assert!(args.get_string_list(&option_id!("dne")).unwrap().is_none());
+
+  let expected_error_msg = "\
+Problem parsing --bad string list value:
+1:['mis', 'matched')
+  -----------------^
+Expected \",\" or the end of a list indicated by ']' at line 1 column 18"
+    .to_owned();
+
+  assert_eq!(
+    expected_error_msg,
+    args.get_string_list(&option_id!("bad")).unwrap_err()
+  );
+}
diff --git a/src/rust/engine/client/src/options/config.rs b/src/rust/engine/client/src/options/config.rs
new file mode 100644
index 00000000000..7d631b53db3
--- /dev/null
+++ b/src/rust/engine/client/src/options/config.rs
@@ -0,0 +1,201 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use std::collections::HashSet;
+use std::fs;
+use std::path::Path;
+
+use toml::value::Table;
+use toml::Value;
+
+use super::id::{NameTransform, OptionId};
+use super::{ListEdit, ListEditAction, OptionsSource};
+
+#[derive(Clone)]
+pub(crate) struct Config {
+  config: Value,
+}
+
+impl Config {
+  pub(crate) fn default() -> Config {
+    Config {
+      config: Value::Table(Table::new()),
+    }
+  }
+
+  pub(crate) fn parse<P: AsRef<Path>>(file: P) -> Result<Config, String> {
+    let config_contents = fs::read_to_string(&file).map_err(|e| {
+      format!(
+        "Failed to read config file {}: {}",
+        file.as_ref().display(),
+        e
+      )
+    })?;
+    let config = config_contents.parse::<Value>().map_err(|e| {
+      format!(
+        "Failed to parse config file {}: {}",
+        file.as_ref().display(),
+        e
+      )
+    })?;
+    if config.is_table() {
+      Ok(Config { config })
+    } else {
+      Err(format!(
+        "Expected the config file {} to contain a table but contained a {}: {}",
+        file.as_ref().display(),
+        config.type_str(),
+        config
+      ))
+    }
+  }
+
+  pub(crate) fn merged<P: AsRef<Path>>(files: &[P]) -> Result<Config, String> {
+    files
+      .iter()
+      .map(Config::parse)
+      .fold(Ok(Config::default()), |acc, parse_result| {
+        acc.and_then(|config| parse_result.map(|parsed| config.merge(parsed)))
+      })
+  }
+
+  fn option_name(id: &OptionId) -> String {
+    id.name("_", NameTransform::None)
+  }
+
+  fn extract_string_list(option_name: &str, value: &Value) -> Result<Vec<String>, String> {
+    if let Some(array) = value.as_array() {
+      let mut items = vec![];
+      for item in array {
+        if let Some(value) = item.as_str() {
+          items.push(value.to_owned())
+        } else {
+          return Err(format!(
+            "Expected {} to be an array of strings but given {} containing non string item {}",
+            option_name, value, item
+          ));
+        }
+      }
+      Ok(items)
+    } else {
+      Err(format!(
+        "Expected {} to be an array but given {}.",
+        option_name, value
+      ))
+    }
+  }
+
+  fn get_value(&self, id: &OptionId) -> Option<&Value> {
+    self
+      .config
+      .get(&id.scope())
+      .and_then(|table| table.get(Self::option_name(id)))
+  }
+
+  pub(crate) fn merge(self, other: Config) -> Config {
+    let mut map = self.config.as_table().unwrap().to_owned();
+    map.extend(
+      other
+        .config
+        .as_table()
+        .unwrap()
+        .iter()
+        .map(|(k, v)| (k.to_owned(), v.to_owned())),
+    );
+    Config {
+      config: Value::Table(map),
+    }
+  }
+}
+
+impl OptionsSource for Config {
+  fn display(&self, id: &OptionId) -> String {
+    format!("{}", id)
+  }
+
+  fn get_string(&self, id: &OptionId) -> Result<Option<String>, String> {
+    if let Some(value) = self.get_value(id) {
+      if let Some(string) = value.as_str() {
+        Ok(Some(string.to_owned()))
+      } else {
+        Err(format!(
+          "Expected {} to be a string but given {}.",
+          id, value
+        ))
+      }
+    } else {
+      Ok(None)
+    }
+  }
+
+  fn get_bool(&self, id: &OptionId) -> Result<Option<bool>, String> {
+    if let Some(value) = self.get_value(id) {
+      if let Some(bool) = value.as_bool() {
+        Ok(Some(bool))
+      } else {
+        Err(format!("Expected {} to be a bool but given {}.", id, value))
+      }
+    } else {
+      Ok(None)
+    }
+  }
+
+  fn get_float(&self, id: &OptionId) -> Result<Option<f64>, String> {
+    if let Some(value) = self.get_value(id) {
+      if let Some(float) = value.as_float() {
+        Ok(Some(float))
+      } else {
+        Err(format!(
+          "Expected {} to be a float but given {}.",
+          id, value
+        ))
+      }
+    } else {
+      Ok(None)
+    }
+  }
+
+  fn get_string_list(&self, id: &OptionId) -> Result<Option<Vec<ListEdit<String>>>, String> {
+    if let Some(table) = self.config.get(&id.scope()) {
+      let option_name = Self::option_name(id);
+      let mut list_edits = vec![];
+      if let Some(value) = table.get(&option_name) {
+        if let Some(sub_table) = value.as_table() {
+          if sub_table.is_empty()
+            || !sub_table.keys().collect::<HashSet<_>>().is_subset(
+              &["add".to_owned(), "remove".to_owned()]
+                .iter()
+                .collect::<HashSet<_>>(),
+            )
+          {
+            return Err(format!(
+              "Expected {} to contain an 'add' element, a 'remove' element or both but found: {:?}",
+              option_name, sub_table
+            ));
+          }
+          if let Some(add) = sub_table.get("add") {
+            list_edits.push(ListEdit {
+              action: ListEditAction::Add,
+              items: Self::extract_string_list(&*format!("{}.add", option_name), &add)?,
+            })
+          }
+          if let Some(remove) = sub_table.get("remove") {
+            list_edits.push(ListEdit {
+              action: ListEditAction::Remove,
+              items: Self::extract_string_list(&*format!("{}.remove", option_name), &remove)?,
+            })
+          }
+        } else {
+          list_edits.push(ListEdit {
+            action: ListEditAction::Replace,
+            items: Self::extract_string_list(&*option_name, value)?,
+          });
+        }
+      }
+      if !list_edits.is_empty() {
+        return Ok(Some(list_edits));
+      }
+    }
+    Ok(None)
+  }
+}
diff --git a/src/rust/engine/client/src/options/env.rs b/src/rust/engine/client/src/options/env.rs
new file mode 100644
index 00000000000..c89246b1237
--- /dev/null
+++ b/src/rust/engine/client/src/options/env.rs
@@ -0,0 +1,75 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use std::collections::HashMap;
+use std::env;
+
+use super::id::{NameTransform, OptionId, Scope};
+use super::OptionsSource;
+use crate::options::parse::{parse_bool, parse_string_list};
+use crate::options::ListEdit;
+
+#[derive(Debug)]
+pub(crate) struct Env {
+  pub(crate) env: HashMap<String, String>,
+}
+
+impl Env {
+  pub(crate) fn capture() -> Env {
+    Env {
+      env: env::vars().collect::<HashMap<_, _>>(),
+    }
+  }
+
+  fn env_var_names(id: &OptionId) -> Vec<String> {
+    let name = id.name("_", NameTransform::ToUpper);
+    let mut names = vec![format!(
+      "PANTS_{}_{}",
+      id.0.name().to_ascii_uppercase(),
+      name
+    )];
+    if id.0 == Scope::Global {
+      names.push(format!("PANTS_{}", name));
+    }
+    if name.starts_with("PANTS_") {
+      names.push(name);
+    }
+    names
+  }
+}
+
+impl OptionsSource for Env {
+  fn display(&self, id: &OptionId) -> String {
+    Self::env_var_names(id).pop().unwrap()
+  }
+
+  fn get_string(&self, id: &OptionId) -> Result<Option<String>, String> {
+    let env_var_names = Self::env_var_names(id);
+    for env_var_name in &env_var_names {
+      if let Some(value) = self.env.get(env_var_name) {
+        return Ok(Some(value.to_owned()));
+      }
+    }
+    Ok(None)
+  }
+
+  fn get_bool(&self, id: &OptionId) -> Result<Option<bool>, String> {
+    if let Some(value) = self.get_string(id)? {
+      parse_bool(&*value)
+        .map(Some)
+        .map_err(|e| e.render(self.display(id)))
+    } else {
+      Ok(None)
+    }
+  }
+
+  fn get_string_list(&self, id: &OptionId) -> Result<Option<Vec<ListEdit<String>>>, String> {
+    if let Some(value) = self.get_string(id)? {
+      parse_string_list(&value)
+        .map(Some)
+        .map_err(|e| e.render(self.display(id)))
+    } else {
+      Ok(None)
+    }
+  }
+}
diff --git a/src/rust/engine/client/src/options/env_tests.rs b/src/rust/engine/client/src/options/env_tests.rs
new file mode 100644
index 00000000000..0187e461afc
--- /dev/null
+++ b/src/rust/engine/client/src/options/env_tests.rs
@@ -0,0 +1,157 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use crate::option_id;
+use crate::options::env::Env;
+use crate::options::{ListEdit, ListEditAction, OptionId, OptionsSource};
+use std::collections::HashMap;
+
+fn env<I: IntoIterator<Item = (&'static str, &'static str)>>(vars: I) -> Env {
+  Env {
+    env: vars
+      .into_iter()
+      .map(|(k, v)| (k.to_owned(), v.to_owned()))
+      .collect::<HashMap<_, _>>(),
+  }
+}
+
+#[test]
+fn test_display() {
+  let env = env([]);
+  assert_eq!(
+    "PANTS_GLOBAL".to_owned(),
+    env.display(&option_id!("global"))
+  );
+  assert_eq!(
+    "PANTS_SCOPE_NAME".to_owned(),
+    env.display(&option_id!("scope", "name"))
+  );
+  assert_eq!(
+    "PANTS_SCOPE_FULL_NAME".to_owned(),
+    env.display(&option_id!(-'f', "scope", "full", "name"))
+  );
+}
+
+#[test]
+fn test_string() {
+  let env = env([
+    ("PANTS_FOO", "bar"),
+    ("PANTS_BAZ_SPAM", "cheese"),
+    ("PANTS_EGGS", "swallow"),
+    ("PANTS_GLOBAL_BOB", "African"),
+    ("PANTS_PANTS_JANE", "elderberry"),
+  ]);
+
+  let assert_string = |expected: &str, id: OptionId| {
+    assert_eq!(expected.to_owned(), env.get_string(&id).unwrap().unwrap())
+  };
+
+  assert_string("bar", option_id!("foo"));
+  assert_string("cheese", option_id!("baz", "spam"));
+  assert_string("swallow", option_id!("pants", "eggs"));
+  assert_string("African", option_id!("bob"));
+  assert_string("elderberry", option_id!("pants", "jane"));
+
+  assert!(env.get_string(&option_id!("dne")).unwrap().is_none());
+}
+
+#[test]
+fn test_bool() {
+  let env = env([
+    ("PANTS_FOO", "true"),
+    ("PANTS_BAR_BAZ", "False"),
+    ("PANTS_EGGS", "swallow"),
+  ]);
+
+  let assert_bool =
+    |expected: bool, id: OptionId| assert_eq!(expected, env.get_bool(&id).unwrap().unwrap());
+
+  assert_bool(true, option_id!("foo"));
+  assert_bool(false, option_id!("bar", "baz"));
+
+  assert!(env.get_bool(&option_id!("dne")).unwrap().is_none());
+  assert_eq!(
+    "Got 'swallow' for PANTS_EGGS. Expected 'true' or 'false'.".to_owned(),
+    env.get_bool(&option_id!("pants", "eggs")).unwrap_err()
+  );
+}
+
+#[test]
+fn test_float() {
+  let env = env([
+    ("PANTS_FOO", "4"),
+    ("PANTS_BAR_BAZ", "3.14"),
+    ("PANTS_EGGS", "1.137"),
+    ("PANTS_BAD", "swallow"),
+  ]);
+
+  let assert_float =
+    |expected: f64, id: OptionId| assert_eq!(expected, env.get_float(&id).unwrap().unwrap());
+
+  assert_float(4 as f64, option_id!("foo"));
+  assert_float(3.14, option_id!("bar", "baz"));
+  assert_float(1.137, option_id!("pants", "eggs"));
+
+  assert!(env.get_float(&option_id!("dne")).unwrap().is_none());
+
+  assert_eq!(
+    "Problem parsing PANTS_BAD value swallow as a float value: invalid float literal".to_owned(),
+    env.get_float(&option_id!("pants", "bad")).unwrap_err()
+  );
+}
+
+#[test]
+fn test_string_list() {
+  let env = env([
+    ("PANTS_BAD", "('mis', 'matched']"),
+    ("PANTS_IMPLICIT_ADD", "initial"),
+    ("PANTS_RESET", "['one']"),
+    ("PANTS_EDITS", "+['two','three'],-['one']"),
+  ]);
+
+  let get_string_list = |id| env.get_string_list(&id).unwrap().unwrap();
+
+  assert_eq!(
+    vec![ListEdit {
+      action: ListEditAction::Add,
+      items: vec!["initial".to_owned()]
+    },],
+    get_string_list(option_id!("implicit", "add"))
+  );
+
+  assert_eq!(
+    vec![ListEdit {
+      action: ListEditAction::Replace,
+      items: vec!["one".to_owned()]
+    },],
+    get_string_list(option_id!("reset"))
+  );
+
+  assert_eq!(
+    vec![
+      ListEdit {
+        action: ListEditAction::Add,
+        items: vec!["two".to_owned(), "three".to_owned()]
+      },
+      ListEdit {
+        action: ListEditAction::Remove,
+        items: vec!["one".to_owned()]
+      },
+    ],
+    get_string_list(option_id!("edits"))
+  );
+
+  assert!(env.get_string_list(&option_id!("dne")).unwrap().is_none());
+
+  let expected_error_msg = "\
+Problem parsing PANTS_BAD string list value:
+1:('mis', 'matched']
+  -----------------^
+Expected \",\" or the end of a tuple indicated by ')' at line 1 column 18"
+    .to_owned();
+
+  assert_eq!(
+    expected_error_msg,
+    env.get_string_list(&option_id!("bad")).unwrap_err()
+  );
+}
diff --git a/src/rust/engine/client/src/options/id.rs b/src/rust/engine/client/src/options/id.rs
new file mode 100644
index 00000000000..f4b9d91395c
--- /dev/null
+++ b/src/rust/engine/client/src/options/id.rs
@@ -0,0 +1,125 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use std::fmt;
+use std::fmt::{Display, Formatter};
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum Scope {
+  Global,
+  Scope(String),
+}
+
+impl Scope {
+  pub fn named(name: &str) -> Scope {
+    match name {
+      "GLOBAL" => Scope::Global,
+      scope => Scope::Scope(scope.to_owned()),
+    }
+  }
+
+  pub fn name(&self) -> &str {
+    match self {
+      Scope::Global => "GLOBAL",
+      Scope::Scope(scope) => scope.as_str(),
+    }
+  }
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct OptionId(
+  pub(crate) Scope,
+  pub(crate) Vec<String>,
+  pub(crate) Option<char>,
+);
+
+impl OptionId {
+  pub fn new<Component, Name>(
+    scope: Scope,
+    name: Name,
+    switch: Option<char>,
+  ) -> Result<OptionId, String>
+  where
+    Component: AsRef<str>,
+    Name: Iterator<Item = Component>,
+  {
+    let name_components = name
+      .map(|component| component.as_ref().to_string())
+      .collect::<Vec<_>>();
+    if name_components.is_empty() {
+      return Err(format!(
+        "Cannot create an OptionId with an empty name. Given a scope of {:?}.",
+        scope
+      ));
+    }
+    Ok(OptionId(scope, name_components, switch))
+  }
+}
+
+impl Display for OptionId {
+  fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+    write!(
+      f,
+      "[{}] {}",
+      self.scope(),
+      self.name("_", NameTransform::None)
+    )
+  }
+}
+
+#[macro_export]
+macro_rules! option_id {
+    (-$switch:literal, [$scope:literal], $($name_component:literal),+) => {
+        $crate::options::OptionId::new(
+            $crate::options::Scope::named($scope),
+            [$($name_component),+].iter(),
+            Some($switch)
+        ).expect("Creating an OptionId via macro should ensure at least one name component")
+    };
+    (-$switch:literal, $($name_component:literal),+) => {
+        $crate::options::OptionId::new(
+            $crate::options::Scope::Global,
+            [$($name_component),+].iter(),
+            Some($switch)
+        ).expect("Creating an OptionId via macro should ensure at least one name component")
+    };
+    ([$scope:literal], $($name_component:literal),+) => {
+        $crate::options::OptionId::new(
+            $crate::options::Scope::named($scope),
+            [$($name_component),+].iter(),
+            None
+        ).expect("Creating an OptionId via macro should ensure at least one name component")
+    };
+    ($($name_component:literal),+) => {
+        $crate::options::OptionId::new(
+            $crate::options::Scope::Global,
+            [$($name_component),+].iter(),
+            None
+        ).expect("Creating an OptionId via macro should ensure at least one name component")
+    };
+}
+
+pub(crate) enum NameTransform {
+  None,
+  ToLower,
+  ToUpper,
+}
+
+impl OptionId {
+  pub(crate) fn scope(&self) -> &str {
+    self.0.name()
+  }
+
+  pub(crate) fn name(&self, sep: &str, transform: NameTransform) -> String {
+    self
+      .1
+      .iter()
+      .map(|component| match transform {
+        NameTransform::None => component.to_owned(),
+        NameTransform::ToLower => component.to_ascii_lowercase(),
+        NameTransform::ToUpper => component.to_ascii_uppercase(),
+      })
+      .collect::<Vec<_>>()
+      .join(sep)
+  }
+}
diff --git a/src/rust/engine/client/src/options/id_tests.rs b/src/rust/engine/client/src/options/id_tests.rs
new file mode 100644
index 00000000000..6516db1e795
--- /dev/null
+++ b/src/rust/engine/client/src/options/id_tests.rs
@@ -0,0 +1,55 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use crate::option_id;
+use crate::options::id::{OptionId, Scope};
+
+#[test]
+fn test_option_id_global_switch() {
+  let option_id = option_id!(-'x', "bar", "baz");
+  assert_eq!(
+    OptionId::new(Scope::Global, ["bar", "baz"].iter(), Some('x')).unwrap(),
+    option_id
+  );
+  assert_eq!("GLOBAL", option_id.scope());
+}
+
+#[test]
+fn test_option_id_global() {
+  let option_id = option_id!("bar", "baz");
+  assert_eq!(
+    OptionId::new(Scope::Global, ["bar", "baz"].iter(), None).unwrap(),
+    option_id
+  );
+  assert_eq!("GLOBAL", option_id.scope());
+}
+
+#[test]
+fn test_option_id_scope_switch() {
+  let option_id = option_id!(-'f', ["foo-bar"], "baz", "spam");
+  assert_eq!(
+    OptionId::new(
+      Scope::Scope("foo-bar".to_owned()),
+      ["baz", "spam"].iter(),
+      Some('f')
+    )
+    .unwrap(),
+    option_id
+  );
+  assert_eq!("foo-bar", option_id.scope());
+}
+
+#[test]
+fn test_option_id_scope() {
+  let option_id = option_id!(["foo-bar"], "baz", "spam");
+  assert_eq!(
+    OptionId::new(
+      Scope::Scope("foo-bar".to_owned()),
+      ["baz", "spam"].iter(),
+      None
+    )
+    .unwrap(),
+    option_id
+  );
+  assert_eq!("foo-bar", option_id.scope());
+}
diff --git a/src/rust/engine/client/src/options/mod.rs b/src/rust/engine/client/src/options/mod.rs
new file mode 100644
index 00000000000..ea5da06e687
--- /dev/null
+++ b/src/rust/engine/client/src/options/mod.rs
@@ -0,0 +1,241 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+mod args;
+#[cfg(test)]
+mod args_tests;
+
+mod config;
+mod env;
+#[cfg(test)]
+mod env_tests;
+
+mod id;
+#[cfg(test)]
+mod id_tests;
+
+mod parse;
+#[cfg(test)]
+mod parse_tests;
+
+use std::collections::{BTreeMap, HashSet};
+use std::ops::Deref;
+use std::os::unix::ffi::OsStrExt;
+use std::path::Path;
+use std::rc::Rc;
+
+use self::args::Args;
+use self::config::Config;
+use self::env::Env;
+pub use self::id::{OptionId, Scope};
+use crate::build_root::BuildRoot;
+use crate::option_id;
+
+#[derive(Copy, Clone, Debug, Eq, PartialEq)]
+pub(crate) enum ListEditAction {
+  Replace,
+  Add,
+  Remove,
+}
+
+#[derive(Debug, Eq, PartialEq)]
+pub(crate) struct ListEdit<T> {
+  pub action: ListEditAction,
+  pub items: Vec<T>,
+}
+
+///
+/// A source of option values.
+///
+/// This is currently a subset of the types of options the Pants python option system handles.
+/// Implementations should mimic the behavior of the equivalent python source.
+///
+pub(crate) trait OptionsSource {
+  ///
+  /// Get a display version of the option `id` that most closely matches the syntax used to supply
+  /// the id at runtime. For example, an global option of "bob" would display as "--bob" for use in
+  /// flag based options and "BOB" in environment variable based options.
+  ///
+  fn display(&self, id: &OptionId) -> String;
+
+  ///
+  /// Get the string option identified by `id` from this source.
+  /// Errors when this source has an option value for `id` but that value is not a string.
+  ///
+  fn get_string(&self, id: &OptionId) -> Result<Option<String>, String>;
+
+  ///
+  /// Get the boolean option identified by `id` from this source.
+  /// Errors when this source has an option value for `id` but that value is not a boolean.
+  ///
+  fn get_bool(&self, id: &OptionId) -> Result<Option<bool>, String>;
+
+  ///
+  /// Get the float option identified by `id` from this source.
+  /// Errors when this source has an option value for `id` but that value is not a float.
+  ///
+  /// The default implementation looks for a string value for `id` and then attempts to parse it as
+  /// a float value.
+  ///
+  fn get_float(&self, id: &OptionId) -> Result<Option<f64>, String> {
+    if let Some(value) = self.get_string(id)? {
+      value.parse().map(Some).map_err(|e| {
+        format!(
+          "Problem parsing {} value {} as a float value: {}",
+          self.display(id),
+          value,
+          e
+        )
+      })
+    } else {
+      Ok(None)
+    }
+  }
+
+  ///
+  /// Get the string list option identified by `id` from this source.
+  /// Errors when this source has an option value for `id` but that value is not a string list.
+  ///
+  fn get_string_list(&self, id: &OptionId) -> Result<Option<Vec<ListEdit<String>>>, String>;
+}
+
+#[derive(Copy, Clone, Debug, Ord, PartialOrd, Eq, PartialEq)]
+pub enum Source {
+  Flag,
+  Env,
+  Config,
+  Default,
+}
+
+#[derive(Debug)]
+pub struct OptionValue<T> {
+  pub source: Source,
+  pub value: T,
+}
+
+impl<T> Deref for OptionValue<T> {
+  type Target = T;
+
+  fn deref(&self) -> &Self::Target {
+    &self.value
+  }
+}
+
+pub struct OptionParser {
+  sources: BTreeMap<Source, Rc<dyn OptionsSource>>,
+}
+
+impl OptionParser {
+  pub fn new() -> Result<OptionParser, String> {
+    let mut sources: BTreeMap<Source, Rc<dyn OptionsSource>> = BTreeMap::new();
+    sources.insert(Source::Env, Rc::new(Env::capture()));
+    sources.insert(Source::Flag, Rc::new(Args::argv()));
+    let mut parser = OptionParser {
+      sources: sources.clone(),
+    };
+
+    let config_path = BuildRoot::find()?.join("pants.toml");
+    let repo_config_files = parser.parse_string_list(
+      &option_id!("pants", "config", "files"),
+      &[
+        std::str::from_utf8(config_path.as_os_str().as_bytes()).map_err(|e| {
+          format!(
+            "Failed to decode build root path {}: {}",
+            config_path.display(),
+            e
+          )
+        })?,
+      ],
+    )?;
+    let mut config = Config::merged(&repo_config_files)?;
+    sources.insert(Source::Config, Rc::new(config.clone()));
+    parser = OptionParser {
+      sources: sources.clone(),
+    };
+
+    if *parser.parse_bool(&option_id!("pantsrc"), true)? {
+      for rcfile in parser.parse_string_list(
+        &option_id!("pantsrc", "files"),
+        &["/etc/pantsrc", shellexpand::tilde("~/.pants.rc").as_ref()],
+      )? {
+        let rcfile_path = Path::new(&rcfile);
+        if rcfile_path.exists() {
+          let rc_config = Config::parse(rcfile_path)?;
+          config = config.merge(rc_config);
+        }
+      }
+    }
+    sources.insert(Source::Config, Rc::new(config));
+    Ok(OptionParser { sources })
+  }
+
+  pub fn parse_bool(&self, id: &OptionId, default: bool) -> Result<OptionValue<bool>, String> {
+    for (source_type, source) in self.sources.iter() {
+      if let Some(value) = source.get_bool(id)? {
+        return Ok(OptionValue {
+          source: *source_type,
+          value,
+        });
+      }
+    }
+    Ok(OptionValue {
+      source: Source::Default,
+      value: default,
+    })
+  }
+
+  pub fn parse_float(&self, id: &OptionId, default: f64) -> Result<OptionValue<f64>, String> {
+    for (source_type, source) in self.sources.iter() {
+      if let Some(value) = source.get_float(id)? {
+        return Ok(OptionValue {
+          source: *source_type,
+          value,
+        });
+      }
+    }
+    Ok(OptionValue {
+      source: Source::Default,
+      value: default,
+    })
+  }
+
+  pub fn parse_string(&self, id: &OptionId, default: &str) -> Result<OptionValue<String>, String> {
+    for (source_type, source) in self.sources.iter() {
+      if let Some(value) = source.get_string(id)? {
+        return Ok(OptionValue {
+          source: *source_type,
+          value,
+        });
+      }
+    }
+    Ok(OptionValue {
+      source: Source::Default,
+      value: default.to_string(),
+    })
+  }
+
+  pub fn parse_string_list(&self, id: &OptionId, default: &[&str]) -> Result<Vec<String>, String> {
+    let mut list_edits = vec![];
+    for (_, source) in self.sources.iter() {
+      if let Some(edits) = source.get_string_list(id)? {
+        list_edits.extend(edits);
+      }
+    }
+    let mut string_list = default.iter().map(|s| s.to_string()).collect::<Vec<_>>();
+    for list_edit in list_edits {
+      match list_edit.action {
+        ListEditAction::Replace => string_list = list_edit.items,
+        ListEditAction::Add => string_list.extend(list_edit.items),
+        ListEditAction::Remove => {
+          let to_remove = list_edit.items.iter().collect::<HashSet<_>>();
+          string_list = string_list
+            .iter()
+            .filter(|item| !to_remove.contains(item))
+            .map(|s| s.to_owned())
+            .collect::<Vec<String>>();
+        }
+      }
+    }
+    Ok(string_list)
+  }
+}
diff --git a/src/rust/engine/client/src/options/parse.rs b/src/rust/engine/client/src/options/parse.rs
new file mode 100644
index 00000000000..e39249060b6
--- /dev/null
+++ b/src/rust/engine/client/src/options/parse.rs
@@ -0,0 +1,206 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use super::{ListEdit, ListEditAction};
+use crate::render_choice;
+
+peg::parser! {
+    grammar option_value_parser() for str {
+        use peg::ParseLiteral;
+
+        rule whitespace() -> ()
+            = quiet!{ " " / "\n" / "\r" / "\t" }
+
+        rule string() -> String
+            = s:(non_escaped_character() / escaped_character())+ { s.into_iter().collect() }
+
+        rule non_escaped_character() -> char
+            = !"\\" c:$([_]) { c.chars().next().unwrap() }
+
+        rule quoted_string() -> String
+            = whitespace()*
+            string:(double_quoted_string() / single_quoted_string())
+            whitespace()* { string }
+
+        rule double_quoted_string() -> String
+            = "\"" s:double_quoted_character()* "\"" { s.into_iter().collect() }
+
+        rule double_quoted_character() -> char
+            = quoted_character("\"")
+            / escaped_character()
+
+        rule single_quoted_string() -> String
+            = "'" s:single_quoted_character()* "'" { s.into_iter().collect() }
+
+        rule single_quoted_character() -> char
+            = quoted_character("'")
+            / escaped_character()
+
+        rule quoted_character(quote_char: &'static str) -> char
+            = !(##parse_string_literal(quote_char) / "\\") c:$([_]) { c.chars().next().unwrap() }
+
+        rule escaped_character() -> char
+            = "\\" c:$([_]) { c.chars().next().unwrap() }
+
+        rule add() -> ListEditAction
+            = "+" { ListEditAction::Add }
+
+        rule remove() -> ListEditAction
+            = "-" { ListEditAction::Remove }
+
+        rule action() -> ListEditAction
+            = quiet!{ action:(add() / remove()) { action } }
+            / expected!(
+                "an optional list edit action of '+' indicating `add` or '-' indicating `remove`"
+            )
+
+        // N.B.: The Python list parsing implementation accepts Python tuple literal syntax too.
+
+        rule tuple_start() -> ()
+            = quiet!{ "(" }
+            / expected!("the start of a tuple indicated by '('")
+
+        rule tuple_end() -> ()
+            = quiet!{ ")" }
+            / expected!("the end of a tuple indicated by ')'")
+
+        rule tuple_items() -> Vec<String>
+            = tuple_start()
+            items:quoted_string() ** ","
+            whitespace()* ","? whitespace()*
+            tuple_end() {
+                items
+            }
+
+        rule list_start() -> ()
+            = quiet!{ "[" }
+            / expected!("the start of a list indicated by '['")
+
+        rule list_end() -> ()
+            = quiet!{ "]" }
+            / expected!("the end of a list indicated by ']'")
+
+        rule list_items() -> Vec<String>
+            = list_start()
+            items:quoted_string() ** ","
+            whitespace()* ","? whitespace()*
+            list_end() {
+                items
+            }
+
+        rule items() -> Vec<String>
+            = whitespace()*
+            items:(tuple_items() / list_items())
+            whitespace()* { items }
+
+        rule list_edit() -> ListEdit<String>
+            = whitespace()* action:action() items:items() whitespace()* {
+                ListEdit { action, items }
+            }
+
+        rule list_edits() -> Vec<ListEdit<String>>
+            = e:list_edit() ** "," ","? { e }
+
+        rule list_replace() -> Vec<ListEdit<String>>
+            = items:items() {
+                vec![ListEdit { action: ListEditAction::Replace, items }]
+            }
+
+        rule implicit_add() -> Vec<ListEdit<String>>
+            = !(whitespace() / add() / remove() / tuple_start() / list_start()) item:string() {
+                vec![ListEdit { action: ListEditAction::Add, items: vec![item.to_owned()] }]
+            }
+
+        pub(crate) rule string_list_edits() -> Vec<ListEdit<String>>
+            = implicit_add() / list_replace() / list_edits()
+    }
+}
+
+mod err {
+  #[derive(Debug, Eq, PartialEq)]
+  pub(crate) struct ParseError {
+    template: String,
+  }
+
+  impl ParseError {
+    pub(super) fn new<S: AsRef<str>>(template: S) -> ParseError {
+      let template_ref = template.as_ref();
+      assert!(
+        template_ref.contains("{name}"),
+        "\
+        Expected the template to contain at least one `{{name}}` placeholder, but found none: \
+        {template}.\
+        ",
+        template = template_ref
+      );
+      ParseError {
+        template: template_ref.to_owned(),
+      }
+    }
+
+    pub(crate) fn render<S: AsRef<str>>(&self, name: S) -> String {
+      self.template.replace("{name}", name.as_ref())
+    }
+  }
+}
+
+pub(crate) use err::ParseError;
+
+fn format_parse_error(
+  type_id: &str,
+  value: &str,
+  parse_error: peg::error::ParseError<peg::str::LineCol>,
+) -> ParseError {
+  let value_with_marker = value
+    .split('\n')
+    .enumerate()
+    .map(|(index, line)| (index + 1, line))
+    .map(|(line_no, line)| {
+      if line_no == parse_error.location.line {
+        format!(
+          "{}:{}\n  {}^",
+          line_no,
+          line,
+          "-".repeat(parse_error.location.column - 1)
+        )
+      } else {
+        format!("{}:{}", line_no, line)
+      }
+    })
+    .collect::<Vec<_>>()
+    .join("\n");
+
+  let mut choices = parse_error.expected.tokens().collect::<Vec<_>>();
+  // N.B.: It appears to be the case that the peg parser parses alternatives concurrently and so
+  // the ordering of choices is observed to be unstable. As such sort them for consistent error
+  // messages.
+  choices.sort_unstable();
+
+  ParseError::new(format!(
+    "\
+    Problem parsing {{name}} {type_id} value:\n{value_with_marker}\nExpected {choices} at \
+    line {line} column {column}\
+    ",
+    type_id = type_id,
+    value_with_marker = value_with_marker,
+    choices = render_choice(choices.as_slice()).unwrap_or_else(|| "nothing".to_owned()),
+    line = parse_error.location.line,
+    column = parse_error.location.column,
+  ))
+}
+
+pub(crate) fn parse_string_list(value: &str) -> Result<Vec<ListEdit<String>>, ParseError> {
+  option_value_parser::string_list_edits(&value)
+    .map_err(|e| format_parse_error("string list", value, e))
+}
+
+pub(crate) fn parse_bool(value: &str) -> Result<bool, ParseError> {
+  match value.to_lowercase().as_str() {
+    "true" => Ok(true),
+    "false" => Ok(false),
+    _ => Err(ParseError::new(format!(
+      "Got '{value}' for {{name}}. Expected 'true' or 'false'.",
+      value = value
+    ))),
+  }
+}
diff --git a/src/rust/engine/client/src/options/parse_tests.rs b/src/rust/engine/client/src/options/parse_tests.rs
new file mode 100644
index 00000000000..5985c09992e
--- /dev/null
+++ b/src/rust/engine/client/src/options/parse_tests.rs
@@ -0,0 +1,209 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use crate::options::parse::{parse_bool, parse_string_list};
+use crate::options::{ListEdit, ListEditAction};
+
+#[test]
+fn test_parse_bool() {
+  assert_eq!(Ok(true), parse_bool("true"));
+  assert_eq!(Ok(true), parse_bool("True"));
+  assert_eq!(Ok(true), parse_bool("TRUE"));
+
+  assert_eq!(Ok(false), parse_bool("false"));
+  assert_eq!(Ok(false), parse_bool("False"));
+  assert_eq!(Ok(false), parse_bool("FALSE"));
+
+  assert_eq!(
+    "Got '1' for foo. Expected 'true' or 'false'.".to_owned(),
+    parse_bool("1").unwrap_err().render("foo")
+  )
+}
+
+#[test]
+fn test_parse_string_list_empty() {
+  assert!(parse_string_list("").unwrap().is_empty());
+}
+
+fn list_edit<I: IntoIterator<Item = &'static str>>(
+  action: ListEditAction,
+  items: I,
+) -> ListEdit<String> {
+  ListEdit {
+    action,
+    items: items.into_iter().map(str::to_owned).collect(),
+  }
+}
+
+const EMPTY_STRING_LIST: [&'static str; 0] = [];
+
+#[test]
+fn test_parse_string_list_replace() {
+  assert_eq!(
+    vec![list_edit(ListEditAction::Replace, EMPTY_STRING_LIST)],
+    parse_string_list("[]").unwrap()
+  );
+  assert_eq!(
+    vec![list_edit(ListEditAction::Replace, ["foo"])],
+    parse_string_list("['foo']").unwrap()
+  );
+  assert_eq!(
+    vec![list_edit(ListEditAction::Replace, ["foo", "bar"])],
+    parse_string_list("['foo','bar']").unwrap()
+  );
+}
+
+#[test]
+fn test_parse_string_list_add() {
+  assert_eq!(
+    vec![list_edit(ListEditAction::Add, EMPTY_STRING_LIST)],
+    parse_string_list("+[]").unwrap()
+  );
+}
+
+#[test]
+fn test_parse_string_list_remove() {
+  assert_eq!(
+    vec![list_edit(ListEditAction::Remove, EMPTY_STRING_LIST)],
+    parse_string_list("-[]").unwrap()
+  );
+}
+
+#[test]
+fn test_parse_string_list_edits() {
+  assert_eq!(
+    vec![
+      list_edit(ListEditAction::Remove, ["foo", "bar"]),
+      list_edit(ListEditAction::Add, ["baz"]),
+      list_edit(ListEditAction::Remove, EMPTY_STRING_LIST),
+    ],
+    parse_string_list("-['foo', 'bar'],+['baz'],-[]").unwrap()
+  );
+}
+
+#[test]
+fn test_parse_string_list_edits_whitespace() {
+  assert_eq!(
+    vec![
+      list_edit(ListEditAction::Remove, ["foo"]),
+      list_edit(ListEditAction::Add, ["bar"]),
+    ],
+    parse_string_list(" - [ 'foo' , ] , + [ 'bar' ] ").unwrap()
+  );
+}
+
+#[test]
+fn test_parse_string_list_implicit_add() {
+  assert_eq!(
+    vec![list_edit(ListEditAction::Add, vec!["foo"])],
+    parse_string_list("foo").unwrap()
+  );
+  assert_eq!(
+    vec![list_edit(ListEditAction::Add, vec!["foo bar"])],
+    parse_string_list("foo bar").unwrap()
+  );
+}
+
+#[test]
+fn test_parse_string_list_quoted_chars() {
+  assert_eq!(
+    vec![list_edit(ListEditAction::Add, vec!["[]"])],
+    parse_string_list(r"\[]").unwrap(),
+    "Expected an implicit add of the literal string `[]` via an escaped opening `[`."
+  );
+  assert_eq!(
+    vec![list_edit(ListEditAction::Add, vec![" "])],
+    parse_string_list(r"\ ").unwrap(),
+    "Expected an implicit add of the literal string ` `."
+  );
+  assert_eq!(
+    vec![list_edit(ListEditAction::Add, vec!["+"])],
+    parse_string_list(r"\+").unwrap(),
+    "Expected an implicit add of the literal string `+`."
+  );
+  assert_eq!(
+    vec![list_edit(ListEditAction::Add, vec!["-"])],
+    parse_string_list(r"\-").unwrap(),
+    "Expected an implicit add of the literal string `-`."
+  );
+  assert_eq!(
+    vec![list_edit(ListEditAction::Replace, vec!["'foo", r"\"])],
+    parse_string_list(r"['\'foo', '\\']").unwrap()
+  );
+}
+
+#[test]
+fn test_parse_string_list_quote_forms() {
+  assert_eq!(
+    vec![list_edit(ListEditAction::Replace, ["foo"])],
+    parse_string_list(r#"["foo"]"#).unwrap(),
+    "Expected double quotes to work."
+  );
+  assert_eq!(
+    vec![list_edit(ListEditAction::Replace, ["foo", "bar"])],
+    parse_string_list(r#"["foo", 'bar']"#).unwrap(),
+    "Expected mixed quote forms to work."
+  );
+}
+
+#[test]
+fn test_parse_string_list_trailing_comma() {
+  assert_eq!(
+    vec![list_edit(ListEditAction::Replace, ["foo"])],
+    parse_string_list("['foo',]").unwrap()
+  );
+  assert_eq!(
+    vec![list_edit(ListEditAction::Replace, ["foo", "bar"])],
+    parse_string_list("['foo','bar',]").unwrap()
+  );
+}
+
+#[test]
+fn test_parse_string_list_whitespace() {
+  assert_eq!(
+    vec![list_edit(ListEditAction::Replace, ["foo"])],
+    parse_string_list(" [ 'foo' ] ").unwrap()
+  );
+  assert_eq!(
+    vec![list_edit(ListEditAction::Replace, ["foo", "bar"])],
+    parse_string_list(" [ 'foo' , 'bar' , ] ").unwrap()
+  );
+}
+
+#[test]
+fn test_parse_string_list_tuple() {
+  assert_eq!(
+    vec![list_edit(ListEditAction::Replace, EMPTY_STRING_LIST)],
+    parse_string_list("()").unwrap()
+  );
+  assert_eq!(
+    vec![list_edit(ListEditAction::Replace, ["foo"])],
+    parse_string_list(r#"("foo")"#).unwrap()
+  );
+  assert_eq!(
+    vec![list_edit(ListEditAction::Replace, ["foo", "bar"])],
+    parse_string_list(r#" ('foo', "bar",)"#).unwrap()
+  );
+}
+
+#[test]
+fn test_parse_string_list_error_formatting() {
+  let bad_input = "\
+-['/etc/hosts'],
+         ?(\"/dev/null\")
+";
+
+  let expected_error_msg = "\
+Problem parsing foo string list value:
+1:-['/etc/hosts'],
+2:         ?(\"/dev/null\")
+  ---------^
+3:
+Expected an optional list edit action of '+' indicating `add` \
+or '-' indicating `remove` at line 2 column 10"
+    .to_owned();
+  assert_eq!(
+    expected_error_msg,
+    parse_string_list(bad_input).unwrap_err().render("foo")
+  )
+}
diff --git a/src/rust/engine/client/src/pantsd.rs b/src/rust/engine/client/src/pantsd.rs
new file mode 100644
index 00000000000..4883805f760
--- /dev/null
+++ b/src/rust/engine/client/src/pantsd.rs
@@ -0,0 +1,171 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use std::path::{Path, PathBuf};
+use std::{fmt, fs};
+
+use libc::pid_t;
+use log::debug;
+use sha2::{Digest, Sha256};
+use sysinfo::{ProcessExt, ProcessStatus, System, SystemExt};
+
+pub(crate) struct Metadata {
+  metadata_dir: PathBuf,
+}
+
+impl Metadata {
+  pub(crate) fn mount<P: AsRef<Path>>(directory: P) -> Result<Metadata, String> {
+    let info = uname::uname().map_err(|e| format!("{}", e))?;
+    let host_hash = Sha256::new()
+      .chain(&info.sysname)
+      .chain(&info.nodename)
+      .chain(&info.release)
+      .chain(&info.version)
+      .chain(&info.machine)
+      .finalize();
+
+    const HOST_FINGERPRINT_LENGTH: usize = 12;
+    let mut hex_digest = String::with_capacity(HOST_FINGERPRINT_LENGTH);
+    for byte in host_hash {
+      fmt::Write::write_fmt(&mut hex_digest, format_args!("{:02x}", byte)).unwrap();
+      if hex_digest.len() >= HOST_FINGERPRINT_LENGTH {
+        break;
+      }
+    }
+
+    let metadata_dir = directory
+      .as_ref()
+      .join(&hex_digest[..HOST_FINGERPRINT_LENGTH])
+      .join("pantsd");
+    if metadata_dir.is_dir() {
+      Ok(Metadata { metadata_dir })
+    } else {
+      Err(format!(
+        "There is no pantsd metadata at {metadata_dir}.",
+        metadata_dir = metadata_dir.display()
+      ))
+    }
+  }
+
+  fn pid(&self) -> Result<pid_t, String> {
+    self
+      .read_metadata("pid")
+      .and_then(|(pid_metadata_path, value)| {
+        value
+          .parse()
+          .map(|pid| {
+            debug!(
+              "Parsed pid {pid} from {pid_metadata_path}.",
+              pid = pid,
+              pid_metadata_path = pid_metadata_path.display()
+            );
+            pid
+          })
+          .map_err(|e| {
+            format!(
+              "Failed to parse pantsd pid from {pid_metadata_path}: {err}",
+              pid_metadata_path = pid_metadata_path.display(),
+              err = e
+            )
+          })
+      })
+  }
+
+  fn process_name(&self) -> Result<String, String> {
+    self.read_metadata("process_name").map(|(_, value)| value)
+  }
+
+  pub(crate) fn port(&self) -> Result<u16, String> {
+    self
+      .read_metadata("socket")
+      .and_then(|(socket_metadata_path, value)| {
+        value
+          .parse()
+          .map(|port| {
+            debug!(
+              "Parsed port {port} from {socket_metadata_path}.",
+              port = port,
+              socket_metadata_path = socket_metadata_path.display()
+            );
+            port
+          })
+          .map_err(|e| {
+            format!(
+              "Failed to parse pantsd port from {socket_metadata_path}: {err}",
+              socket_metadata_path = &socket_metadata_path.display(),
+              err = e
+            )
+          })
+      })
+  }
+
+  fn read_metadata(&self, name: &str) -> Result<(PathBuf, String), String> {
+    let metadata_path = self.metadata_dir.join(name);
+    fs::read_to_string(&metadata_path)
+      .map_err(|e| {
+        format!(
+          "Failed to read {name} from {metadata_path}: {err}",
+          name = name,
+          metadata_path = &metadata_path.display(),
+          err = e
+        )
+      })
+      .map(|value| (metadata_path, value))
+  }
+}
+
+pub fn probe(working_dir: &Path, metadata_dir: &Path) -> Result<u16, String> {
+  let pantsd_metadata = Metadata::mount(metadata_dir)?;
+
+  // Grab the purported port early. If we can't get that, then none of the following checks
+  // are useful.
+  let port = pantsd_metadata.port()?;
+
+  let pid = pantsd_metadata.pid()?;
+  let mut system = System::new();
+  system.refresh_process(pid);
+  // Check that the recorded pid is a live process.
+  match system.get_process(pid) {
+    None => {
+      return Err(format!(
+        "\
+        The last pid for the pantsd controlling {working_dir} was {pid} but it no longer appears \
+        to be running.\
+        ",
+        working_dir = working_dir.display(),
+        pid = pid,
+      ))
+    }
+    Some(process) => {
+      // Check that the live process is in fact the expected pantsd process (i.e.: pids have not
+      // wrapped).
+      if std::mem::discriminant(&ProcessStatus::Zombie) == std::mem::discriminant(&process.status())
+      {
+        return Err(format!("The pantsd at pid {pid} is a zombie.", pid = pid));
+      }
+      let expected_process_name_prefix = pantsd_metadata.process_name()?;
+      let actual_argv0 = {
+        let actual_command_line = process.cmd();
+        if actual_command_line.is_empty() {
+          process.name()
+        } else {
+          &actual_command_line[0]
+        }
+      };
+      // It appears the the daemon only records a prefix of the process name, so we just check that.
+      if actual_argv0.starts_with(&expected_process_name_prefix) {
+        Ok(port)
+      } else {
+        Err(format!(
+          "\
+          The process with pid {pid} is not pantsd. Expected a process name matching \
+          {expected_name} but is {actual_name}.\
+          ",
+          pid = pid,
+          expected_name = expected_process_name_prefix,
+          actual_name = actual_argv0
+        ))
+      }
+    }
+  }
+}
diff --git a/src/rust/engine/client/src/pantsd_testing.rs b/src/rust/engine/client/src/pantsd_testing.rs
new file mode 100644
index 00000000000..fbbf33d4f0e
--- /dev/null
+++ b/src/rust/engine/client/src/pantsd_testing.rs
@@ -0,0 +1,51 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use std::fs;
+use std::process::{Command, Stdio};
+use std::str::from_utf8;
+
+use tempdir::TempDir;
+
+use crate::build_root::BuildRoot;
+
+pub(crate) fn launch_pantsd() -> (BuildRoot, TempDir) {
+  let build_root = BuildRoot::find()
+    .expect("Expected test to be run inside the Pants repo but no build root was detected.");
+  let pants_subprocessdir = TempDir::new("pants_subproccessdir").unwrap();
+  let mut cmd = Command::new(build_root.join("pants"));
+  cmd
+    .current_dir(build_root.as_path())
+    .arg("--pants-config-files=[]")
+    .arg("--no-pantsrc")
+    .arg("--pantsd")
+    .arg(format!(
+      "--pants-subprocessdir={}",
+      pants_subprocessdir.path().display()
+    ))
+    .arg("-V")
+    .stderr(Stdio::inherit());
+  let result = cmd
+    .output()
+    .map_err(|e| {
+      format!(
+        "Problem running command {command:?}: {err}",
+        command = cmd,
+        err = e
+      )
+    })
+    .unwrap();
+  assert_eq!(Some(0), result.status.code());
+  assert_eq!(
+    fs::read_to_string(
+      build_root
+        .join("src")
+        .join("python")
+        .join("pants")
+        .join("VERSION")
+    )
+    .unwrap(),
+    from_utf8(result.stdout.as_slice()).unwrap()
+  );
+  (build_root, pants_subprocessdir)
+}
diff --git a/src/rust/engine/client/src/pantsd_tests.rs b/src/rust/engine/client/src/pantsd_tests.rs
new file mode 100644
index 00000000000..b8495de8799
--- /dev/null
+++ b/src/rust/engine/client/src/pantsd_tests.rs
@@ -0,0 +1,34 @@
+// Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
+// Licensed under the Apache License, Version 2.0 (see LICENSE).
+
+use std::net::TcpStream;
+
+use crate::pantsd;
+use crate::pantsd_testing::launch_pantsd;
+
+fn assert_connect(port: u16) {
+  assert!(
+    port >= 1024,
+    "Pantsd should never be running on a privileged port."
+  );
+
+  let stream = TcpStream::connect(("0.0.0.0", port)).unwrap();
+  assert_eq!(port, stream.peer_addr().unwrap().port());
+}
+
+#[test]
+fn test_address_integration() {
+  let (_, pants_subprocessdir) = launch_pantsd();
+
+  let pantsd_metadata = pantsd::Metadata::mount(&pants_subprocessdir).unwrap();
+  let port = pantsd_metadata.port().unwrap();
+  assert_connect(port);
+}
+
+#[test]
+fn test_probe() {
+  let (build_root, pants_subprocessdir) = launch_pantsd();
+
+  let port = pantsd::probe(&build_root, pants_subprocessdir.path()).unwrap();
+  assert_connect(port);
+}