diff --git a/CHANGELOG.md b/CHANGELOG.md
index fb33523..1f3e055 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - `--print-build-logs` / `-L` argument.
+- `nix copy` & `nix build` operations are now manually implemented through a native
+  rust nix daemon client.
+- Node `target` liveliness is now determined directly by initiating a nix daemon
+  handshake.
 
 ### Changed
 
diff --git a/Cargo.lock b/Cargo.lock
index 06dc409..9c71ec1 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3041,6 +3041,7 @@ dependencies = [
  "tokio-util",
  "tracing",
  "wire-key-agent",
+ "wire-nix-client",
  "zstd",
 ]
 
@@ -3059,6 +3060,20 @@ dependencies = [
  "tokio-util",
 ]
 
+[[package]]
+name = "wire-nix-client"
+version = "1.3.0"
+dependencies = [
+ "itertools",
+ "miette",
+ "nix-compat",
+ "owo-colors",
+ "serde",
+ "thiserror 2.0.18",
+ "tokio",
+ "tracing",
+]
+
 [[package]]
 name = "wit-bindgen"
 version = "0.51.0"
diff --git a/Cargo.toml b/Cargo.toml
index eba5698..4925a3a 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,5 +1,5 @@
 [workspace]
-members = ["crates/key_agent", "crates/core", "crates/cli"]
+members = ["crates/key_agent", "crates/core", "crates/cli", "crates/nix_client"]
 resolver = "2"
 package.edition = "2024"
 package.version = "1.3.0"
@@ -17,6 +17,7 @@ missing_errors_doc = "allow"
 missing_panics_doc = "allow"
 
 [workspace.dependencies]
+itertools = "0.14.0"
 futures-util = { version = "0.3.31", features = ["sink", "std"] }
 clap = { version = "4.5.51", features = ["derive", "string", "cargo"] }
 clap-verbosity-flag = "3.0.4"
@@ -37,6 +38,7 @@ base64 = "0.22.1"
 nix-compat = { git = "https://git.snix.dev/snix/snix.git", features = [
   "serde",
   "flakeref",
+  "daemon",
 ] }
 # simd-json = { version = "0.17.0", features = [
 #   "serde_impl",
diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml
index 99a2dbc..2ed2add 100644
--- a/crates/core/Cargo.toml
+++ b/crates/core/Cargo.toml
@@ -27,7 +27,7 @@ rand = "0.10.0"
 tokio-util = { workspace = true }
 portable-pty = "0.9.0"
 anyhow.workspace = true
-itertools = "0.14.0"
+itertools = { workspace = true }
 enum_dispatch = "0.3.13"
 sha2 = { workspace = true }
 base64 = { workspace = true }
@@ -38,6 +38,7 @@ owo-colors = { workspace = true }
 termion = "4.0.6"
 sqlx = { version = "0.9", features = ["runtime-tokio", "sqlite"] }
 zstd = "0.13.3"
+wire-nix-client = { path = "../nix_client" }
 
 [dev-dependencies]
 tempdir = "0.3"
diff --git a/crates/core/src/commands/common.rs b/crates/core/src/commands/common.rs
index ea001b2..f4f380e 100644
--- a/crates/core/src/commands/common.rs
+++ b/crates/core/src/commands/common.rs
@@ -1,88 +1,17 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 // Copyright 2024-2025 wire Contributors
 
-use std::collections::HashMap;
-
 use tracing::instrument;
 
 use crate::{
     EvalGoal, SubCommandModifiers,
     commands::{
         CommandArguments, Either, WireCommandChip, builder::CommandStringBuilder, run_command,
-        run_command_with_env,
     },
     errors::{CommandError, HiveInitialisationError, HiveLibError},
-    hive::{
-        HiveLocation,
-        node::{Context, Push, SharedTarget},
-    },
+    hive::HiveLocation,
 };
 
-fn get_common_copy_path_help(error: &CommandError) -> Option<String> {
-    if let CommandError::CommandFailed { logs, .. } = error
-        && (logs.contains("error: unexpected end-of-file"))
-    {
-        Some("wire requires the deploying user or wire binary cache is trusted on the remote server. if you're attempting to make that change, skip keys with --no-keys. please read https://wire.forall.systems/guides/keys for more information".to_string())
-    } else {
-        None
-    }
-}
-
-pub async fn push(
-    context: &Context,
-    target: &SharedTarget,
-    push: Push<'_>,
-    substitute_on_destination: bool,
-) -> Result<(), HiveLibError> {
-    let target = target.0.read().await;
-
-    let mut command_string = CommandStringBuilder::nix();
-
-    command_string.args(&["--extra-experimental-features", "nix-command", "copy"]);
-    command_string.opt_arg(substitute_on_destination, "--substitute-on-destination");
-    command_string.arg("--to");
-    command_string.args(&[
-        format!(
-            "ssh://{user}@{host}",
-            user = target.user,
-            host = target.get_preferred_host()?,
-        ),
-        match push {
-            Push::Derivation(drv) => format!("{}^* --derivation", drv.to_absolute_path()),
-            Push::Path(path) => path.to_absolute_path(),
-        },
-    ]);
-
-    let child = run_command_with_env(
-        &CommandArguments::new(command_string, context.modifiers)
-            .mode(crate::commands::ChildOutputMode::Nix),
-        HashMap::from([(
-            "NIX_SSHOPTS".into(),
-            target.create_ssh_opts(context.modifiers)?,
-        )]),
-    )
-    .await?;
-
-    let status = child.wait_till_success().await;
-
-    let help = if let Err(ref error) = status {
-        get_common_copy_path_help(error).map(Box::new)
-    } else {
-        None
-    };
-
-    status.map_err(|error| HiveLibError::NixCopyError {
-        name: context.name.clone(),
-        path: match push {
-            Push::Derivation(path) | Push::Path(path) => path.clone(),
-        },
-        error: Box::new(error),
-        help,
-    })?;
-
-    Ok(())
-}
-
 fn get_common_command_help(error: &CommandError) -> Option<String> {
     if let CommandError::CommandFailed { logs, .. } = error
         // marshmallow: your using this repo as a hive you idiot
diff --git a/crates/core/src/commands/mod.rs b/crates/core/src/commands/mod.rs
index 91e13d2..70d06d1 100644
--- a/crates/core/src/commands/mod.rs
+++ b/crates/core/src/commands/mod.rs
@@ -33,7 +33,6 @@ pub(crate) mod pty;
 pub(crate) enum ChildOutputMode {
     Nix,
     Generic,
-    Interactive,
 }
 
 #[derive(Debug)]
@@ -158,6 +157,86 @@ impl WireCommandChip for Either<InteractiveChildChip, NonInteractiveChildChip> {
     }
 }
 
+pub(crate) fn trace_nix_log_message(
+    log_message: LogMessage,
+    build_name_map: &BuildNameMap,
+    print_build_logs: bool,
+) -> Option<String> {
+    let (msg, level, build_name) = match log_message {
+        LogMessage::Start {
+            text,
+            r#type,
+            level,
+            id,
+            fields,
+            ..
+        } => {
+            if let Some(fields) = fields
+                && matches!(r#type, nix_compat::log::ActivityType::Build)
+                // first field of start log contains the build name.
+                && let Some(Field::String(name)) = fields.first()
+                && print_build_logs
+            {
+                build_name_map
+                    .lock()
+                    .insert(id, drv_path_to_build_name(name));
+            }
+
+            (text, level, None)
+        }
+        LogMessage::Stop { id, .. } => {
+            build_name_map.lock().remove(&id);
+
+            return None;
+        }
+        LogMessage::Msg { msg, level, .. } => (msg, level, None),
+        LogMessage::Result {
+            r#type: ResultType::BuildLogLine,
+            fields,
+            id,
+            ..
+        } => {
+            if !print_build_logs {
+                return None;
+            }
+
+            let Some(Field::String(msg)) = fields.into_iter().next() else {
+                return None;
+            };
+
+            // Attempt to reuse owned bytes into a utf8 string, or falls
+            // back lossy if it fails.
+            let msg = match msg {
+                std::borrow::Cow::Borrowed(bytes) => String::from_utf8_lossy(bytes),
+                std::borrow::Cow::Owned(vec) => match String::from_utf8(vec) {
+                    Ok(s) => Cow::Owned(s),
+                    Err(e) => Cow::Owned(String::from_utf8_lossy(e.as_bytes()).into_owned()),
+                },
+            };
+
+            let lock = build_name_map.lock();
+            let build_name = lock.get(&id).cloned();
+
+            (msg, VerbosityLevel::Info, build_name)
+        }
+        _ => return None,
+    };
+
+    if msg.is_empty() {
+        return None;
+    }
+
+    let msg = strip_ansi_escapes::strip_str(msg);
+
+    let level = log_print(&level, build_name.as_ref(), &msg);
+
+    if matches!(level, tracing::Level::ERROR | tracing::Level::WARN) {
+        return Some(msg);
+    }
+
+    None
+}
+
 impl ChildOutputMode {
     /// this function is by far the biggest hotspot in the whole tree
     /// Returns a string if this log is notable to be stored as an error message
@@ -168,7 +247,7 @@ impl ChildOutputMode {
         print_build_logs: bool,
     ) -> Option<String> {
         let slice = match self {
-            Self::Generic | Self::Interactive => {
+            Self::Generic => {
                 let string = String::from_utf8_lossy(line);
                 let stripped = strip_ansi_escapes::strip_str(&string);
                 warn!("{stripped}");
@@ -193,79 +272,7 @@ impl ChildOutputMode {
             return None;
         };
 
-        let (msg, level, build_name) = match log_message {
-            LogMessage::Start {
-                text,
-                r#type,
-                level,
-                id,
-                fields,
-                ..
-            } => {
-                if let Some(fields) = fields
-                    && matches!(r#type, nix_compat::log::ActivityType::Build)
-                    // first field of start log contains the build name.
-                    && let Some(Field::String(name)) = fields.first()
-                    && print_build_logs
-                {
-                    build_name_map
-                        .lock()
-                        .insert(id, drv_path_to_build_name(name));
-                }
-
-                (text, level, None)
-            }
-            LogMessage::Stop { id, .. } => {
-                build_name_map.lock().remove(&id);
-
-                return None;
-            }
-            LogMessage::Msg { msg, level, .. } => (msg, level, None),
-            LogMessage::Result {
-                r#type: ResultType::BuildLogLine,
-                fields,
-                id,
-                ..
-            } => {
-                if !print_build_logs {
-                    return None;
-                }
-
-                let Some(Field::String(msg)) = fields.into_iter().next() else {
-                    return None;
-                };
-
-                // Attempt to reuse owned bytes into a utf8 string, or falls
-                // back lossy if it fails.
-                let msg = match msg {
-                    std::borrow::Cow::Borrowed(bytes) => String::from_utf8_lossy(bytes),
-                    std::borrow::Cow::Owned(vec) => match String::from_utf8(vec) {
-                        Ok(s) => Cow::Owned(s),
-                        Err(e) => Cow::Owned(String::from_utf8_lossy(e.as_bytes()).into_owned()),
-                    },
-                };
-
-                let lock = build_name_map.lock();
-                let build_name = lock.get(&id).cloned();
-
-                (msg, VerbosityLevel::Info, build_name)
-            }
-            _ => return None,
-        };
-
-        if msg.is_empty() {
-            return None;
-        }
-
-        let msg = strip_ansi_escapes::strip_str(msg);
-
-        let level = log_print(&level, build_name.as_ref(), &msg);
-
-        if matches!(level, tracing::Level::ERROR | tracing::Level::WARN) {
-            return Some(msg);
-        }
-
-        None
+        trace_nix_log_message(log_message, build_name_map, print_build_logs)
     }
 }
 
diff --git a/crates/core/src/commands/noninteractive.rs b/crates/core/src/commands/noninteractive.rs
index 8925e41..7cb1e30 100644
--- a/crates/core/src/commands/noninteractive.rs
+++ b/crates/core/src/commands/noninteractive.rs
@@ -50,7 +50,7 @@ pub(crate) async fn non_interactive_command_with_env<S: AsRef<str>>(
         "{command_string}{extra}",
         command_string = arguments.command_string.as_ref(),
         extra = match arguments.output_mode {
-            ChildOutputMode::Generic | ChildOutputMode::Interactive => "",
+            ChildOutputMode::Generic => "",
             ChildOutputMode::Nix => " --log-format internal-json",
         }
     );
@@ -201,7 +201,7 @@ async fn create_sync_ssh_command(
 ) -> Result<Command, HiveLibError> {
     let target = target.0.read().await;
     let mut command = Command::new("ssh");
-    command.args(target.create_ssh_args(modifiers)?);
+    command.args(target.create_ssh_args(modifiers, false)?);
     command.arg(target.get_preferred_host()?.to_string());
 
     Ok(command)
diff --git a/crates/core/src/commands/pty/mod.rs b/crates/core/src/commands/pty/mod.rs
index 2455f6d..feb3387 100644
--- a/crates/core/src/commands/pty/mod.rs
+++ b/crates/core/src/commands/pty/mod.rs
@@ -83,43 +83,21 @@ static FAILED_PATTERN: LazyLock<PatternID> = LazyLock::new(|| PatternID::must(2)
 /// substitutes STDOUT with #$line. stdout is far less common than stderr.
 const IO_SUBS: &str = "1> >(while IFS= read -r line; do echo \"#$line\"; done)";
 
-fn create_ending_segment<S: AsRef<str>>(
-    arguments: &CommandArguments<S>,
-    needles: &Needles,
-) -> String {
-    let Needles {
-        succeed,
-        fail,
-        start,
-    } = needles;
+fn create_ending_segment(needles: &Needles) -> String {
+    let Needles { succeed, fail, .. } = needles;
 
     format!(
         "echo -e '{succeed}' || echo '{failed}'",
-        succeed = if matches!(arguments.output_mode, ChildOutputMode::Interactive) {
-            format!(
-                "{start}\\n{succeed}",
-                start = String::from_utf8_lossy(start),
-                succeed = String::from_utf8_lossy(succeed)
-            )
-        } else {
-            String::from_utf8_lossy(succeed).to_string()
-        },
+        succeed = String::from_utf8_lossy(succeed),
         failed = String::from_utf8_lossy(fail)
     )
 }
 
-fn create_starting_segment<S: AsRef<str>>(
-    arguments: &CommandArguments<S>,
-    start_needle: &Arc<Vec<u8>>,
-) -> String {
-    if matches!(arguments.output_mode, ChildOutputMode::Interactive) {
-        String::new()
-    } else {
-        format!(
-            "echo '{start}' && ",
-            start = String::from_utf8_lossy(start_needle)
-        )
-    }
+fn create_starting_segment(start_needle: &Arc<Vec<u8>>) -> String {
+    format!(
+        "echo '{start}' && ",
+        start = String::from_utf8_lossy(start_needle)
+    )
 }
 
 #[instrument(skip_all, name = "run-int", fields(elevated = %arguments.is_elevated(), mode = ?arguments.output_mode))]
@@ -139,10 +117,10 @@ pub(crate) async fn interactive_command_with_env<S: AsRef<str>>(
         command = arguments.command_string.as_ref(),
         flags = match arguments.output_mode {
             ChildOutputMode::Nix => "--log-format internal-json",
-            ChildOutputMode::Generic | ChildOutputMode::Interactive => "",
+            ChildOutputMode::Generic => "",
         },
-        starting = create_starting_segment(arguments, &needles.start),
-        ending = create_ending_segment(arguments, &needles)
+        starting = create_starting_segment(&needles.start),
+        ending = create_ending_segment(&needles)
     );
 
     debug!("{command_string}");
@@ -449,7 +427,7 @@ async fn create_int_ssh_command(
 ) -> Result<portable_pty::CommandBuilder, HiveLibError> {
     let target = target.0.read().await;
     let mut command = portable_pty::CommandBuilder::new("ssh");
-    command.args(target.create_ssh_args(modifiers)?);
+    command.args(target.create_ssh_args(modifiers, false)?);
     command.arg(target.get_preferred_host()?.to_string());
     Ok(command)
 }
diff --git a/crates/core/src/errors.rs b/crates/core/src/errors.rs
index 172803f..3d52432 100644
--- a/crates/core/src/errors.rs
+++ b/crates/core/src/errors.rs
@@ -9,6 +9,7 @@ use miette::{Diagnostic, SourceSpan};
 use nix_compat::flakeref::{FlakeRef, FlakeRefError};
 use thiserror::Error;
 use tokio::task::JoinError;
+use wire_nix_client::{NixDaemonClientError, store_path::StorePathError};
 
 use crate::{
     SafeStorePath,
@@ -233,6 +234,25 @@ pub enum HiveLibError {
     #[diagnostic(transparent)]
     HiveLocationError(HiveLocationError),
 
+    #[error(transparent)]
+    #[diagnostic(transparent)]
+    NixDaemonClientError(NixDaemonClientError),
+
+    #[error(transparent)]
+    #[diagnostic(transparent)]
+    StorePath(StorePathError),
+
+    #[diagnostic(code(wire::CopyPath))]
+    #[error("failed to copy path {} to node {name}", path.to_absolute_path())]
+    NixCopyError {
+        name: Name,
+        path: SafeStorePath<String>,
+        #[source]
+        error: Box<NixDaemonClientError>,
+        #[help]
+        help: Option<String>,
+    },
+
     #[error("Failed to apply key {}", .0)]
     KeyError(
         String,
@@ -246,18 +266,7 @@ pub enum HiveLibError {
     NixBuildError {
         name: Name,
         #[source]
-        source: CommandError,
-    },
-
-    #[diagnostic(code(wire::CopyPath))]
-    #[error("failed to copy path {} to node {name}", path.to_absolute_path())]
-    NixCopyError {
-        name: Name,
-        path: SafeStorePath<String>,
-        #[source]
-        error: Box<CommandError>,
-        #[help]
-        help: Option<Box<String>>,
+        source: NixDaemonClientError,
     },
 
     #[diagnostic(code(wire::Evaluate))]
@@ -279,12 +288,4 @@ pub enum HiveLibError {
     #[diagnostic(code(wire::SIGINT))]
     #[error("SIGINT received, shut down")]
     Sigint,
-
-    #[diagnostic(code(wire::SnixStorePath))]
-    #[error("Failed to parse store path {path:?}")]
-    StorePath {
-        path: String,
-        #[source]
-        error: nix_compat::store_path::Error,
-    },
 }
diff --git a/crates/core/src/hive/executor.rs b/crates/core/src/hive/executor.rs
index ec9b3ce..caf496c 100644
--- a/crates/core/src/hive/executor.rs
+++ b/crates/core/src/hive/executor.rs
@@ -52,6 +52,7 @@ async fn evaluate_task(
             debug!(pre_parsed_output = %output, "evaluated {name}");
 
             SafeStorePath::<String>::from_absolute_path(output.as_bytes())
+                .map_err(HiveLibError::StorePath)
         });
 
     debug!(output = ?output, done = true);
diff --git a/crates/core/src/hive/node.rs b/crates/core/src/hive/node.rs
index 7af175e..55ff91b 100644
--- a/crates/core/src/hive/node.rs
+++ b/crates/core/src/hive/node.rs
@@ -13,8 +13,7 @@ use std::sync::nonpoison::Mutex;
 use tokio::sync::{RwLock, oneshot};
 use tracing::instrument;
 
-use crate::commands::builder::CommandStringBuilder;
-use crate::commands::{CommandArguments, WireCommandChip, run_command};
+use crate::commands::trace_nix_log_message;
 use crate::errors::NetworkError;
 use crate::hive::HiveLocation;
 use crate::hive::steps::build::Build;
@@ -22,7 +21,7 @@ use crate::hive::steps::evaluate::Evaluate;
 use crate::hive::steps::keys::{Key, Keys, PushKeyAgent};
 use crate::hive::steps::ping::Ping;
 use crate::hive::steps::push::{PushBuildOutput, PushEvaluatedOutput};
-use crate::{SafeStorePath, StrictHostKeyChecking, SubCommandModifiers};
+use crate::{SafeStorePath, StrictHostKeyChecking, SubCommandModifiers, open_remote_client};
 
 use super::HiveLibError;
 use super::steps::activate::SwitchToConfiguration;
@@ -66,13 +65,14 @@ impl PartialEq for SharedTarget {
 impl Target {
     #[instrument(ret(level = tracing::Level::DEBUG), skip_all)]
     pub fn create_ssh_opts(&self, modifiers: SubCommandModifiers) -> Result<String, HiveLibError> {
-        self.create_ssh_args(modifiers).map(|x| x.join(" "))
+        self.create_ssh_args(modifiers, false).map(|x| x.join(" "))
     }
 
     #[instrument(ret(level = tracing::Level::DEBUG))]
     pub fn create_ssh_args(
         &self,
         modifiers: SubCommandModifiers,
+        force_quiet: bool,
     ) -> Result<Vec<String>, HiveLibError> {
         let mut vector = vec![
             "-l".to_string(),
@@ -96,7 +96,9 @@ impl Target {
         vector.push("-o".to_string());
         vector.extend(options.into_iter().intersperse("-o".to_string()));
 
-        if modifiers.ssh_verbosity > 0 {
+        if force_quiet {
+            vector.push("-q".to_string());
+        } else if modifiers.ssh_verbosity > 0 {
             vector.push(format!("-{}", "v".repeat(modifiers.ssh_verbosity)));
         }
 
@@ -104,27 +106,14 @@ impl Target {
     }
 
     /// Tests the connection to a node
-    pub async fn ping(&self, modifiers: SubCommandModifiers) -> Result<(), HiveLibError> {
-        let host = self.get_preferred_host()?;
-
-        let mut command_string = CommandStringBuilder::new("ssh");
-        command_string.arg(format!("{}@{host}", self.user));
-        command_string.arg(self.create_ssh_opts(modifiers)?);
-        command_string.arg("exit");
-
-        let output = run_command(
-            &CommandArguments::new(command_string, modifiers)
-                .log_stdout()
-                .mode(crate::commands::ChildOutputMode::Interactive),
-        )
-        .await?;
-
-        output.wait_till_success().await.map_err(|source| {
-            HiveLibError::NetworkError(NetworkError::HostUnreachable {
-                host: host.to_string(),
-                source,
-            })
-        })?;
+    pub async fn ping(
+        &self,
+        modifiers: SubCommandModifiers,
+        should_quit: Arc<AtomicBool>,
+    ) -> Result<(), HiveLibError> {
+        open_remote_client(&self, modifiers, trace_nix_log_message, should_quit).await?;
+
+        // connection established
 
         Ok(())
     }
@@ -235,6 +224,7 @@ pub fn should_apply_locally(allow_local_deployment: bool, name: &str) -> bool {
     *name == *gethostname() && allow_local_deployment
 }
 
+#[derive(Debug)]
 pub enum Push<'a> {
     Derivation(&'a SafeStorePath<String>),
     Path(&'a SafeStorePath<String>),
@@ -353,14 +343,17 @@ mod tests {
             "BatchMode=yes".to_string(),
         ];
 
-        assert_eq!(target.create_ssh_args(subcommand_modifiers).unwrap(), args);
+        assert_eq!(
+            target.create_ssh_args(subcommand_modifiers, false).unwrap(),
+            args
+        );
         assert_eq!(
             target.create_ssh_opts(subcommand_modifiers).unwrap(),
             args.join(" ")
         );
 
         assert_eq!(
-            target.create_ssh_args(subcommand_modifiers).unwrap(),
+            target.create_ssh_args(subcommand_modifiers, false).unwrap(),
             [
                 "-l".to_string(),
                 target.user.to_string(),
@@ -374,7 +367,7 @@ mod tests {
         );
 
         assert_eq!(
-            target.create_ssh_args(subcommand_modifiers).unwrap(),
+            target.create_ssh_args(subcommand_modifiers, false).unwrap(),
             [
                 "-l".to_string(),
                 target.user.to_string(),
@@ -389,12 +382,15 @@ mod tests {
 
         // forced non interactive is the same as --non-interactive
         assert_eq!(
-            target.create_ssh_args(subcommand_modifiers).unwrap(),
+            target.create_ssh_args(subcommand_modifiers, false).unwrap(),
             target
-                .create_ssh_args(SubCommandModifiers {
-                    non_interactive: true,
-                    ..Default::default()
-                })
+                .create_ssh_args(
+                    SubCommandModifiers {
+                        non_interactive: true,
+                        ..Default::default()
+                    },
+                    false
+                )
                 .unwrap()
         );
     }
diff --git a/crates/core/src/hive/steps/activate.rs b/crates/core/src/hive/steps/activate.rs
index 52bb499..3671b7e 100644
--- a/crates/core/src/hive/steps/activate.rs
+++ b/crates/core/src/hive/steps/activate.rs
@@ -34,7 +34,7 @@ async fn wait_for_ping(target: &SharedTarget, ctx: &Context) -> Result<(), HiveL
     for num in 0..3 {
         warn!("Trying to ping {host} (attempt {}/3)", num + 1);
 
-        let result = target.ping(ctx.modifiers).await;
+        let result = target.ping(ctx.modifiers, ctx.should_quit.clone()).await;
 
         if result.is_ok() {
             info!("Regained connection to {} via {host}", ctx.name);
diff --git a/crates/core/src/hive/steps/build.rs b/crates/core/src/hive/steps/build.rs
index 9e6b50b..5e4dee5 100644
--- a/crates/core/src/hive/steps/build.rs
+++ b/crates/core/src/hive/steps/build.rs
@@ -3,17 +3,18 @@
 
 use std::fmt::Display;
 
-use tracing::{info, instrument};
+use tracing::{debug, info, instrument};
+use wire_nix_client::{DerivedPath, DerivedPathOutput, NixClient, NixDaemonClientError};
 
 use crate::{
-    HiveLibError, SafeStorePath,
-    commands::{
-        CommandArguments, Either, WireCommandChip, builder::CommandStringBuilder,
-        run_command_with_env,
-    },
+    HiveLibError,
+    commands::{Either, trace_nix_log_message},
     hive::node::{Context, ExecuteStep, SharedTarget},
+    open_remote_client,
 };
 
+const SYSTEM_OUTPUT: &str = "out";
+
 #[derive(Debug)]
 #[cfg_attr(test, derive(PartialEq))]
 pub struct Build {
@@ -31,45 +32,73 @@ impl ExecuteStep for Build {
     async fn execute(&self, ctx: &mut Context) -> Result<(), HiveLibError> {
         let top_level = ctx.state.evaluation.as_ref().unwrap();
 
-        let mut command_string = CommandStringBuilder::nix();
-        command_string.args(&[
-            "--extra-experimental-features",
-            "nix-command",
-            "build",
-            "--no-link",
-            "--print-out-paths",
-        ]);
-        command_string.opt_arg(ctx.modifiers.print_build_logs, "--print-build-logs");
-        command_string.arg(format!("{}^*", top_level.to_absolute_path()));
-
-        let status = run_command_with_env(
-            &CommandArguments::new(command_string, ctx.modifiers)
-                // build remotely if asked for AND we arent applying locally
-                .execute_on_remote(self.target.clone())
-                .mode(crate::commands::ChildOutputMode::Nix)
-                .log_stdout(),
-            std::collections::HashMap::new(),
-        )
-        .await?
-        .wait_till_success()
-        .await
-        .map_err(|source| HiveLibError::NixBuildError {
+        let mut connection = if let Some(ref target) = self.target {
+            let target = target.0.read().await;
+
+            Either::Left(
+                open_remote_client(
+                    &target,
+                    ctx.modifiers,
+                    trace_nix_log_message,
+                    ctx.should_quit.clone(),
+                )
+                .await?
+                .0,
+            )
+        } else {
+            Either::Right(
+                NixClient::open_local(
+                    trace_nix_log_message,
+                    ctx.should_quit.clone(),
+                    ctx.modifiers.print_build_logs,
+                )
+                .await
+                .map_err(HiveLibError::NixDaemonClientError)?,
+            )
+        };
+
+        let mut output_map = match connection {
+            Either::Left(ref mut conn) => conn.query_derivation_output_map(top_level).await,
+            Either::Right(ref mut conn) => conn.query_derivation_output_map(top_level).await,
+        }
+        .map_err(|err| HiveLibError::NixBuildError {
             name: ctx.name.clone(),
-            source,
+            source: err,
         })?;
 
-        let stdout = match status {
-            Either::Left((_, stdout)) | Either::Right((_, stdout)) => stdout,
+        debug!(output_map = ?output_map, "got output map");
+
+        let output_path =
+            output_map
+                .remove(SYSTEM_OUTPUT)
+                .flatten()
+                .ok_or(HiveLibError::NixBuildError {
+                    name: ctx.name.clone(),
+                    source: NixDaemonClientError::NixDaemonInvalidResponse(format!(
+                        "Derivation {top_level:?} did not have output {SYSTEM_OUTPUT:?}"
+                    )),
+                })?;
+
+        let derived_path = DerivedPath {
+            store_path: top_level,
+            outputs: DerivedPathOutput::OutputNames(&[SYSTEM_OUTPUT]),
         };
 
-        info!("Built output: {stdout:?}");
+        match connection {
+            Either::Left(mut conn) => conn.build(&vec![derived_path]).await,
+            Either::Right(mut conn) => conn.build(&vec![derived_path]).await,
+        }
+        .map_err(|source| HiveLibError::NixBuildError {
+            name: ctx.name.clone(),
+            source,
+        })?;
+
+        info!("Built output: {output_path:?}");
 
         // print built path to stdout
-        println!("{stdout}");
+        println!("{}", output_path.to_absolute_path());
 
-        ctx.state.build = Some(SafeStorePath::<String>::from_absolute_path(
-            stdout.as_bytes(),
-        )?);
+        ctx.state.build = Some(output_path);
 
         Ok(())
     }
diff --git a/crates/core/src/hive/steps/keys.rs b/crates/core/src/hive/steps/keys.rs
index b45cdaa..01681f0 100644
--- a/crates/core/src/hive/steps/keys.rs
+++ b/crates/core/src/hive/steps/keys.rs
@@ -27,11 +27,10 @@ use tokio_util::codec::LengthDelimitedCodec;
 use tracing::{debug, instrument};
 
 use crate::commands::builder::CommandStringBuilder;
-use crate::commands::common::push;
 use crate::commands::{CommandArguments, WireCommandChip, run_command};
 use crate::errors::KeyError;
 use crate::hive::node::{Context, ExecuteStep, Push, SharedTarget};
-use crate::{HiveLibError, SafeStorePath};
+use crate::{HiveLibError, SafeStorePath, push};
 
 #[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq, Hash)]
 #[serde(tag = "t", content = "c")]
@@ -324,7 +323,8 @@ impl ExecuteStep for PushKeyAgent {
         };
 
         let agent_store_path =
-            SafeStorePath::<String>::from_absolute_path(agent_directory.as_bytes())?;
+            SafeStorePath::<String>::from_absolute_path(agent_directory.as_bytes())
+                .map_err(HiveLibError::StorePath)?;
 
         if let Some(ref target) = self.target {
             push(
diff --git a/crates/core/src/hive/steps/ping.rs b/crates/core/src/hive/steps/ping.rs
index ddda6f1..7ec6130 100644
--- a/crates/core/src/hive/steps/ping.rs
+++ b/crates/core/src/hive/steps/ping.rs
@@ -34,7 +34,11 @@ impl ExecuteStep for Ping {
                 host = target.get_preferred_host()?.to_string()
             );
 
-            if target.ping(ctx.modifiers).await.is_ok() {
+            if target
+                .ping(ctx.modifiers, ctx.should_quit.clone())
+                .await
+                .is_ok()
+            {
                 event!(
                     Level::INFO,
                     status = "success",
diff --git a/crates/core/src/hive/steps/push.rs b/crates/core/src/hive/steps/push.rs
index 760f1a1..e88cad3 100644
--- a/crates/core/src/hive/steps/push.rs
+++ b/crates/core/src/hive/steps/push.rs
@@ -7,8 +7,8 @@ use tracing::instrument;
 
 use crate::{
     HiveLibError,
-    commands::common::push,
     hive::node::{Context, ExecuteStep, SharedTarget},
+    push,
 };
 
 #[derive(Debug)]
diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs
index 22daa9c..b41f3c1 100644
--- a/crates/core/src/lib.rs
+++ b/crates/core/src/lib.rs
@@ -5,14 +5,28 @@
 #![feature(sync_nonpoison)]
 #![feature(nonpoison_mutex)]
 
-use std::{io::IsTerminal, sync::LazyLock};
+use std::{
+    collections::{HashMap, HashSet},
+    io::IsTerminal,
+    ops::Deref,
+    process::Stdio,
+    sync::{Arc, LazyLock, atomic::AtomicBool, nonpoison::Mutex},
+};
 
-use serde::Deserialize;
-use tokio::sync::{AcquireError, Semaphore, SemaphorePermit, mpsc::UnboundedSender, oneshot};
+use nix_compat::log::LogMessage;
+use tokio::{
+    process::{ChildStdin, ChildStdout, Command},
+    sync::{AcquireError, Semaphore, SemaphorePermit, mpsc::UnboundedSender, oneshot},
+};
+use tracing::{info, instrument, trace};
+use wire_nix_client::{
+    NixClient, NixDaemonClientError, WireAddToStoreNarRequest, store_path::SafeStorePath,
+};
 
 use crate::{
+    commands::trace_nix_log_message,
     errors::HiveLibError,
-    hive::node::Name,
+    hive::node::{Context, Name, Push, SharedTarget},
     status::{UI_SENDER, UiMessage},
 };
 
@@ -99,89 +113,178 @@ pub async fn acquire_stdin_lock<'a>() -> Result<ClobberGuard<'a>, AcquireError>
     Ok(result)
 }
 
-/// This type exists to restrict `StorePath` usage to only methods that deal with
-/// absolute paths. By default, the `StorePath` type implements Display that
-/// does not include `/nix/store/` can introduce many hard to catch bugs.
-///
-///
-/// If <https://github.com/rust-lang/rust-clippy/issues/8581>
-/// is ever closed, this can be dropped from the codebase.
-#[derive(Debug, Clone)]
-#[allow(clippy::disallowed_types)]
-pub struct SafeStorePath<S>(nix_compat::store_path::StorePath<S>);
-
-#[allow(clippy::disallowed_types)]
-impl<S> SafeStorePath<S>
+#[instrument(skip(trace_callback))]
+pub async fn open_remote_client<D, T>(
+    target: &D,
+    modifiers: SubCommandModifiers,
+    trace_callback: T,
+    should_quit: Arc<AtomicBool>,
+) -> Result<(NixClient<ChildStdout, ChildStdin, T>, String), HiveLibError>
 where
-    S: AsRef<str>,
+    D: Deref<Target = crate::hive::node::Target> + std::fmt::Debug,
+    T: Fn(LogMessage, &Arc<Mutex<HashMap<u64, Arc<String>>>>, bool) -> Option<String>,
 {
-    pub fn from_absolute_path<'a>(s: &'a [u8]) -> Result<SafeStorePath<S>, HiveLibError>
-    where
-        S: From<&'a str>,
-    {
-        Ok(Self(
-            nix_compat::store_path::StorePath::from_absolute_path(s).map_err(|error| {
-                HiveLibError::StorePath {
-                    path: String::from_utf8_lossy(s).to_string(),
-                    error,
-                }
-            })?,
-        ))
-    }
+    let mut command = Command::new("ssh")
+        .args(target.create_ssh_args(modifiers, true)?)
+        .arg(target.get_preferred_host()?.to_string())
+        .arg("nix-daemon --stdio")
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        // TODO: move to separate thread
+        .stderr(Stdio::inherit())
+        .spawn()
+        .map_err(|error| {
+            HiveLibError::NixDaemonClientError(NixDaemonClientError::NixDaemonConnectionFailure(
+                error,
+            ))
+        })?;
+
+    let stdin = command.stdin.take().unwrap();
+    let stdout = command.stdout.take().unwrap();
+
+    tokio::spawn(async move { command.wait().await });
+
+    Ok((
+        NixClient::<ChildStdout, ChildStdin, T>::handshake(
+            stdout,
+            stdin,
+            trace_callback,
+            should_quit,
+            modifiers.print_build_logs,
+        )
+        .await
+        .map_err(HiveLibError::NixDaemonClientError)?,
+        target.get_preferred_host()?.to_string(),
+    ))
+}
 
-    pub fn from_name_and_digest<'a>(name: &'a str, digest: &[u8]) -> Result<Self, HiveLibError>
-    where
-        S: From<&'a str>,
+fn get_common_copy_path_help(error: &NixDaemonClientError) -> Option<String> {
+    if let NixDaemonClientError::NixDaemonOperationError { msg, .. } = error
+        && (msg.contains("error: unexpected end-of-file"))
     {
-        Ok(Self(
-            nix_compat::store_path::StorePath::from_name_and_digest(name, digest).map_err(
-                |error| HiveLibError::StorePath {
-                    path: format!("raw name & digest: {digest:?}-{name:?}"),
-                    error,
-                },
-            )?,
-        ))
+        Some("wire requires the deploying user or wire binary cache is trusted on the remote server. if you're attempting to make that change, skip keys with --no-keys. please read https://wire.forall.systems/guides/keys for more information".to_string())
+    } else {
+        None
     }
+}
 
-    pub fn into_inner(self) -> nix_compat::store_path::StorePath<S> {
-        self.0
-    }
+pub async fn push(
+    context: &Context,
+    target: &SharedTarget,
+    push: Push<'_>,
+    substitute_on_destination: bool,
+) -> Result<(), HiveLibError> {
+    let mut local_daemon = NixClient::open_local(
+        trace_nix_log_message,
+        context.should_quit.clone(),
+        context.modifiers.print_build_logs,
+    )
+    .await
+    .map_err(HiveLibError::NixDaemonClientError)?;
 
-    pub fn to_absolute_path(&self) -> String {
-        self.0.to_absolute_path()
-    }
+    let target = target.0.read().await;
 
-    pub fn digest(&self) -> &[u8; nix_compat::store_path::DIGEST_SIZE] {
-        self.0.digest()
-    }
+    let (mut remote_daemon, host) = open_remote_client(
+        &target,
+        context.modifiers,
+        trace_nix_log_message,
+        context.should_quit.clone(),
+    )
+    .await?;
 
-    pub fn name(&self) -> &S {
-        self.0.name()
-    }
-}
+    let path = match push {
+        Push::Derivation(path) | Push::Path(path) => path.clone(),
+    };
 
-#[allow(clippy::disallowed_types)]
-impl<'de, S> Deserialize<'de> for SafeStorePath<S>
-where
-    nix_compat::store_path::StorePath<S>: Deserialize<'de>,
-{
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-    where
-        D: serde::Deserializer<'de>,
-    {
-        Ok(SafeStorePath(
-            nix_compat::store_path::StorePath::deserialize(deserializer)?,
-        ))
+    info!(path = ?path, "attempting copy");
+
+    let closure = local_daemon
+        .collect_complete_closure(&path)
+        .await
+        .map_err(HiveLibError::NixDaemonClientError)?;
+    let closure_length = closure.len();
+
+    info!(path = ?path, "closure has {:?} paths", closure_length);
+
+    let paths_on_target: HashSet<_> = remote_daemon
+        .query_valid_paths(closure.clone(), substitute_on_destination)
+        .await
+        .map_err(HiveLibError::NixDaemonClientError)?
+        .into_iter()
+        .collect();
+
+    trace!(path = ?path, "target already has {} path(s)", paths_on_target.len());
+
+    let paths_to_push = closure_length.saturating_sub(paths_on_target.len());
+    if paths_to_push > 0 {
+        info!("pushing {}", closure_length - paths_on_target.len());
     }
-}
 
-impl<S> PartialEq for SafeStorePath<S>
-where
-    S: AsRef<str>,
-{
-    fn eq(&self, other: &Self) -> bool {
-        self.0.eq(&other.0)
+    let paths_to_upload = closure.into_iter().filter(|p| !paths_on_target.contains(p));
+
+    for path in paths_to_upload {
+        info!("copying '{}' to node {host}", path.to_absolute_path());
+
+        let Some(path_info) =
+            local_daemon
+                .query(&path)
+                .await
+                .map_err(|err| HiveLibError::NixCopyError {
+                    name: context.name.clone(),
+                    path: path.clone(),
+                    help: get_common_copy_path_help(&err),
+                    error: Box::new(err),
+                })?
+        else {
+            return Err(HiveLibError::NixCopyError {
+                name: context.name.clone(),
+                path: path.clone(),
+                error: Box::new(NixDaemonClientError::NixDaemonOperationFailed(format!(
+                    "selected {path:?} for upload does not exist in local store"
+                ))),
+                help: None,
+            });
+        };
+
+        let nar_stream = local_daemon
+            .get_nar_stream(&path, path_info.nar_size)
+            .await
+            .map_err(|err| HiveLibError::NixCopyError {
+                name: context.name.clone(),
+                path: path.clone(),
+                help: get_common_copy_path_help(&err),
+                error: Box::new(err),
+            })?;
+
+        remote_daemon
+            .add_to_store_nar(
+                WireAddToStoreNarRequest {
+                    path: path.clone(),
+                    deriver: path_info.deriver.map(Into::into),
+                    nar_hash: path_info.nar_hash,
+                    references: path_info
+                        .references
+                        .into_iter()
+                        .map(SafeStorePath)
+                        .collect(),
+                    registration_time: path_info.registration_time,
+                    nar_size: path_info.nar_size,
+                    ultimate: false,
+                    signatures: path_info.signatures,
+                    ca: path_info.ca,
+                    repair: false,
+                    dont_check_sigs: true,
+                },
+                nar_stream,
+            )
+            .await
+            .map_err(|err| HiveLibError::NixCopyError {
+                name: context.name.clone(),
+                path: path.clone(),
+                help: get_common_copy_path_help(&err),
+                error: Box::new(err),
+            })?;
     }
-}
 
-impl<S> Eq for SafeStorePath<S> where S: AsRef<str> {}
+    Ok(())
+}
diff --git a/crates/nix_client/Cargo.toml b/crates/nix_client/Cargo.toml
new file mode 100644
index 0000000..8332669
--- /dev/null
+++ b/crates/nix_client/Cargo.toml
@@ -0,0 +1,17 @@
+[package]
+name = "wire-nix-client"
+edition.workspace = true
+version.workspace = true
+
+[dependencies]
+serde = { workspace = true }
+itertools = { workspace = true }
+nix-compat = { workspace = true }
+owo-colors = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+miette = { workspace = true }
+thiserror = { workspace = true }
+
+[lints]
+workspace = true
diff --git a/crates/nix_client/src/lib.rs b/crates/nix_client/src/lib.rs
new file mode 100644
index 0000000..90fcf21
--- /dev/null
+++ b/crates/nix_client/src/lib.rs
@@ -0,0 +1,1010 @@
+#![feature(sync_nonpoison)]
+#![feature(nonpoison_mutex)]
+
+use std::sync::atomic::AtomicBool;
+use std::{
+    borrow::{Borrow, Cow},
+    collections::{BTreeMap, HashMap, HashSet},
+    sync::{Arc, nonpoison::Mutex},
+};
+
+use itertools::Itertools;
+use miette::Diagnostic;
+use nix_compat::{
+    log::VerbosityLevel,
+    narinfo::Signature,
+    nix_daemon::types::UnkeyedValidPathInfo,
+    nixhash::CAHash,
+    wire::{
+        ProtocolVersion,
+        de::{NixRead, NixReader, NixReaderBuilder},
+        read_string,
+        ser::{NixWrite, NixWriter, NixWriterBuilder},
+    },
+    worker_protocol::Operation,
+};
+use nix_compat::{
+    log::{ActivityType, Field, LogMessage, ResultType},
+    wire::{de::NixDeserialize, ser::NixSerialize},
+};
+use owo_colors::{OwoColorize, Stream};
+use thiserror::Error;
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt, ReadHalf, Take, WriteHalf, split},
+    net::UnixStream,
+};
+use tracing::{Level, debug, info, instrument, trace, warn};
+
+use crate::store_path::SafeStorePath;
+
+pub mod store_path;
+
+// https://snix.dev/docs/reference/nix-daemon-protocol/changelog/
+const CLIENT_VERSION: ProtocolVersion = ProtocolVersion::from_parts(1, 37);
+const CLIENT_ONE: u64 = 0x6e69_7863;
+const SERVER_ONE: u64 = 0x6478_696f;
+
+const STDERR_ERROR: u64 = 0x6378_7470; // "cxtp"
+const STDERR_READ: u64 = 0x6461_7461; // "data"
+const STDERR_LAST: u64 = 0x616c_7473;
+const STDERR_NEXT: u64 = 0x6f6c_6d67;
+const STDERR_WRITE: u64 = 0x6461_7416;
+const STDERR_START_ACTIVITY: u64 = 0x5354_5254;
+const STDERR_STOP_ACTIVITY: u64 = 0x5354_4f50;
+const STDERR_RESULT: u64 = 0x5253_4c54;
+
+#[derive(Debug, Diagnostic, Error)]
+pub enum NixDaemonClientError {
+    #[diagnostic(code(wire::NixDaemonIO))]
+    #[error("nix daemon io error")]
+    NixDaemonIO(#[source] std::io::Error),
+
+    #[diagnostic(code(wire::NixDaemonInvalidResponse))]
+    #[error("nix daemon returned an invalid response: {}", .0)]
+    NixDaemonInvalidResponse(String),
+
+    #[diagnostic(code(wire::NixDaemonOperationFailed))]
+    #[error("nix daemon operation failed: {}", .0)]
+    NixDaemonOperationFailed(String),
+
+    #[diagnostic(code(wire::NixDaemonConnectionFailure))]
+    #[error("failed to connect to nix daemon")]
+    NixDaemonConnectionFailure(#[source] std::io::Error),
+
+    #[diagnostic(code(wire::NixDaemonProtocolVersion))]
+    #[error(
+        "the nix daemon protocol version is too old for wire to perform {operation:?}! want atleast {wanted}, have {have}"
+    )]
+    NixDaemonProtocolVersion {
+        wanted: nix_compat::wire::ProtocolVersion,
+        have: nix_compat::wire::ProtocolVersion,
+        operation: String,
+    },
+
+    #[diagnostic(code(wire::NixDaemonOperationError))]
+    #[error("{name}: {msg}")]
+    NixDaemonOperationError { name: String, msg: String },
+
+    #[diagnostic(code(wire::SIGINT))]
+    #[error("SIGINT received, shut down")]
+    Sigint,
+}
+
+// `AddToStoreNar` with SafeSTorePath & a nar_hash which is a string instead of
+// a CAHash (which cannot be easily deserialized and serialised)
+#[derive(Debug)]
+pub struct WireAddToStoreNarRequest {
+    pub path: SafeStorePath<String>,
+    pub deriver: Option<SafeStorePath<String>>,
+    pub nar_hash: String,
+    pub references: Vec<SafeStorePath<String>>,
+    pub registration_time: u64,
+    pub nar_size: u64,
+    pub ultimate: bool,
+    pub signatures: Vec<Signature<String>>,
+    pub ca: Option<CAHash>,
+    pub repair: bool,
+    pub dont_check_sigs: bool,
+}
+
+#[derive(Debug)]
+pub struct QueryMissingResult {
+    will_build: HashSet<SafeStorePath<String>>,
+    will_substitute: HashSet<SafeStorePath<String>>,
+    _unknown: HashSet<SafeStorePath<String>>,
+    download_size: u64,
+    _nar_size: u64,
+}
+
+#[derive(Debug)]
+pub enum DerivedPathOutput<'a, S: Borrow<str> + std::fmt::Debug> {
+    // wont support `*` as wire does not require that yet and some protocols
+    // dont have it.
+    // /// All (*) outputs
+    // All,
+    /// List of output names
+    OutputNames(&'a [S]),
+}
+
+#[derive(Debug)]
+pub struct DerivedPath<'a, S: Borrow<str> + std::fmt::Debug> {
+    pub store_path: &'a SafeStorePath<String>,
+    pub outputs: DerivedPathOutput<'a, S>,
+}
+
+pub struct NixClient<R, W, T> {
+    reader: NixReader<R>,
+    writer: NixWriter<W>,
+
+    build_name_map: Arc<Mutex<HashMap<u64, Arc<String>>>>,
+    should_quit: Arc<AtomicBool>,
+    print_build_logs: bool,
+
+    trace_callback: T,
+}
+
+impl<T> NixClient<UnixStream, UnixStream, T>
+where
+    T: Fn(LogMessage, &Arc<Mutex<HashMap<u64, Arc<String>>>>, bool) -> Option<String>,
+{
+    #[instrument(skip(trace_callback))]
+    pub async fn open_local(
+        trace_callback: T,
+        should_quit: Arc<AtomicBool>,
+        print_build_logs: bool,
+    ) -> Result<NixClient<ReadHalf<UnixStream>, WriteHalf<UnixStream>, T>, NixDaemonClientError>
+    {
+        let stream = UnixStream::connect("/nix/var/nix/daemon-socket/socket")
+            .await
+            .map_err(NixDaemonClientError::NixDaemonConnectionFailure)?;
+
+        let (reader, writer) = split(stream);
+
+        NixClient::<ReadHalf<UnixStream>, WriteHalf<UnixStream>, T>::handshake(
+            reader,
+            writer,
+            trace_callback,
+            should_quit,
+            print_build_logs,
+        )
+        .await
+    }
+}
+
+impl<R, W, T> NixClient<R, W, T>
+where
+    T: Fn(LogMessage, &Arc<Mutex<HashMap<u64, Arc<String>>>>, bool) -> Option<String>,
+{
+    #[instrument(skip_all)]
+    pub async fn handshake(
+        mut reader: R,
+        mut writer: W,
+        trace_callback: T,
+        should_quit: Arc<AtomicBool>,
+        print_build_logs: bool,
+    ) -> Result<NixClient<R, W, T>, NixDaemonClientError>
+    where
+        R: AsyncReadExt + std::fmt::Debug + Unpin + Send,
+        W: AsyncWriteExt + std::fmt::Debug + Unpin + Send,
+    {
+        trace!("sending {CLIENT_ONE:x?} in handshake");
+
+        writer
+            .write_u64_le(CLIENT_ONE)
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)?;
+
+        let magic = reader
+            .read_u64_le()
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)?;
+
+        trace!("server responded with magic {magic:x?}");
+
+        if magic != SERVER_ONE {
+            return Err(NixDaemonClientError::NixDaemonInvalidResponse(format!(
+                "daemon returned invalid magic in handshake: {magic:?}"
+            )));
+        }
+
+        let protocol_version: u64 = reader
+            .read_u64_le()
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)?;
+
+        trace!(server_version = ?protocol_version);
+
+        let server_version: nix_compat::wire::ProtocolVersion =
+            protocol_version.try_into().map_err(|error: &str| {
+                NixDaemonClientError::NixDaemonInvalidResponse(error.to_string())
+            })?;
+
+        // send our version
+        writer
+            .write_u64_le(u64::from(CLIENT_VERSION))
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)?;
+
+        // sendCpu, hardcoded to false
+        writer
+            .write_u8(0)
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)?;
+
+        // reserveSpace, obsolete
+        writer
+            .write_u8(0)
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)?;
+
+        let nix_version = read_string(&mut reader, 0..=10)
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)?;
+        let trusted = reader
+            .read_u8()
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)?;
+
+        debug!(daemon_nix_version = ?nix_version, server_version = ?server_version, trusted = ?trusted, "completed handshake with daemon");
+
+        let reader = NixReaderBuilder::default()
+            .set_version(server_version)
+            .build(reader);
+        let writer = NixWriterBuilder::default()
+            .set_version(server_version)
+            .build(writer);
+
+        let mut result = Self {
+            reader,
+            writer,
+            build_name_map: Arc::new(Mutex::new(HashMap::new())),
+            trace_callback,
+            should_quit,
+            print_build_logs,
+        };
+
+        result.drain_stderr().await?;
+
+        return Ok(result);
+    }
+
+    #[instrument(level = Level::TRACE, skip_all, ret)]
+    async fn read_value<V>(&mut self) -> Result<V, NixDaemonClientError>
+    where
+        R: AsyncReadExt + std::fmt::Debug + Unpin + Send,
+        V: NixDeserialize + std::fmt::Debug,
+    {
+        self.shutdown_guard()?;
+
+        self.reader
+            .read_value::<V>()
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)
+    }
+
+    #[instrument(level = Level::TRACE, skip(self), ret)]
+    async fn write_value<V>(&mut self, value: &V) -> Result<(), NixDaemonClientError>
+    where
+        W: AsyncWriteExt + std::fmt::Debug + Unpin + Send,
+        V: NixSerialize + std::fmt::Debug + Send,
+    {
+        self.shutdown_guard()?;
+
+        self.writer
+            .write_value(value)
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)
+    }
+
+    #[instrument(skip(self))]
+    async fn read_error(&mut self) -> Result<NixDaemonClientError, NixDaemonClientError>
+    where
+        R: AsyncReadExt + std::fmt::Debug + Unpin + Send,
+    {
+        let _type: String = self.read_value().await?;
+        let _level: VerbosityLevel = self.read_value().await?;
+        let name: String = self.read_value().await?;
+        let msg: String = self.read_value().await?;
+        let _have_pos: u64 = self.read_value().await?;
+
+        Ok(NixDaemonClientError::NixDaemonOperationError { name, msg })
+    }
+
+    fn shutdown_guard(&self) -> Result<(), NixDaemonClientError> {
+        if self.should_quit.load(std::sync::atomic::Ordering::Relaxed) {
+            return Err(NixDaemonClientError::Sigint);
+        }
+
+        Ok(())
+    }
+
+    #[instrument(skip(self))]
+    async fn drain_stderr(&mut self) -> Result<(), NixDaemonClientError>
+    where
+        R: AsyncReadExt + std::fmt::Debug + Unpin + Send,
+        W: AsyncWriteExt + std::fmt::Debug + Unpin + Send,
+    {
+        loop {
+            self.shutdown_guard()?;
+
+            let msg_type: u64 = self.read_value().await?;
+
+            match msg_type {
+                STDERR_LAST => {
+                    trace!("stderr stream ended normally");
+                    break;
+                }
+                STDERR_ERROR => {
+                    debug!("stderr error encountered");
+                    return Err(self.read_error().await?);
+                }
+                STDERR_NEXT => {
+                    let msg = self.read_value().await?;
+
+                    // normal string log message
+                    (self.trace_callback)(
+                        LogMessage::Msg {
+                            level: VerbosityLevel::Info,
+                            msg: Cow::Owned(msg),
+                        },
+                        &self.build_name_map,
+                        self.print_build_logs,
+                    );
+                }
+                STDERR_START_ACTIVITY => {
+                    let activity = self.read_activity_start().await?;
+
+                    (self.trace_callback)(activity, &self.build_name_map, self.print_build_logs);
+                }
+                STDERR_STOP_ACTIVITY => {
+                    let id = self.read_value().await?;
+                    (self.trace_callback)(
+                        LogMessage::Stop { id },
+                        &self.build_name_map,
+                        self.print_build_logs,
+                    );
+                }
+                STDERR_RESULT => {
+                    let id: u64 = self.read_value().await?;
+                    let result_type: ResultType = ResultType::try_from(
+                        u8::try_from(self.read_value::<u64>().await?).map_err(|err| {
+                            NixDaemonClientError::NixDaemonInvalidResponse(format!(
+                                "could not cast result type to u8: {err:?}"
+                            ))
+                        })?,
+                    )
+                    .map_err(|err| {
+                        NixDaemonClientError::NixDaemonInvalidResponse(format!(
+                            "could not convert u64 to ResultType: {err:?}"
+                        ))
+                    })?;
+
+                    let fields = self.read_activity_fields().await?;
+
+                    (self.trace_callback)(
+                        LogMessage::Result {
+                            fields,
+                            id,
+                            r#type: result_type,
+                        },
+                        &self.build_name_map,
+                        self.print_build_logs,
+                    );
+                }
+                STDERR_READ => {
+                    let _desired_len: u64 = self.read_value().await?;
+
+                    warn!("STDERR_READ is not implemented")
+                }
+                STDERR_WRITE => {
+                    // todo: read bytes here
+                    warn!("STDERR_WRITE is not implemented")
+                }
+                _ => {
+                    return Err(NixDaemonClientError::NixDaemonInvalidResponse(format!(
+                        "unknown daemon message type in stderr stream: {msg_type:x}"
+                    )));
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    // https://snix.dev/docs/reference/nix-daemon-protocol/types/#field
+    #[instrument(skip(self))]
+    async fn read_activity_start<'a>(&mut self) -> Result<LogMessage<'a>, NixDaemonClientError>
+    where
+        R: AsyncReadExt + std::fmt::Debug + Unpin + Send,
+        W: AsyncWriteExt + std::fmt::Debug + Unpin + Send,
+    {
+        let id: u64 = self.read_value().await?;
+        let level: VerbosityLevel = self.read_value().await?;
+
+        let activity_type: ActivityType = ActivityType::try_from(
+            u8::try_from(self.read_value::<u64>().await?).map_err(|err| {
+                NixDaemonClientError::NixDaemonInvalidResponse(format!(
+                    "could not cast activity type to u8: {err:?}"
+                ))
+            })?,
+        )
+        .map_err(|err| {
+            NixDaemonClientError::NixDaemonInvalidResponse(format!(
+                "could not convert u64 to ActivityType: {err:?}"
+            ))
+        })?;
+
+        let text: String = self.read_value().await?;
+
+        let fields = self.read_activity_fields().await?;
+
+        let parent: u64 = self.read_value().await?;
+
+        return Ok(LogMessage::Start {
+            fields: Some(fields),
+            id,
+            level,
+            parent,
+            text: text.into(),
+            r#type: activity_type,
+        });
+    }
+
+    // https://snix.dev/docs/reference/nix-daemon-protocol/types/#field
+    #[instrument(skip(self))]
+    async fn read_activity_fields<'a>(&mut self) -> Result<Vec<Field<'a>>, NixDaemonClientError>
+    where
+        R: AsyncReadExt + std::fmt::Debug + Unpin + Send,
+        W: AsyncWriteExt + std::fmt::Debug + Unpin + Send,
+    {
+        let fields_count: u64 = self.read_value().await?;
+
+        let mut fields = Vec::new();
+
+        for _ in 0..fields_count {
+            let field_type: u64 = self.read_value().await?;
+
+            match field_type {
+                0 => {
+                    fields.push(Field::Int(self.read_value().await?));
+                }
+                1 => {
+                    let str_val: String = self.read_value().await?;
+
+                    fields.push(Field::String(Cow::Owned(str_val.into_bytes())));
+                }
+                _ => {
+                    return Err(NixDaemonClientError::NixDaemonInvalidResponse(format!(
+                        "unknown activity field type: {field_type}",
+                    )));
+                }
+            }
+        }
+
+        Ok(fields)
+    }
+
+    /// Takes a list of store paths and returns a new list only containing the valid store paths
+    /// more information: <https://snix.dev/docs/reference/nix-daemon-protocol/operations/#queryvalidpaths>
+    #[instrument(skip_all)]
+    pub async fn query_valid_paths(
+        &mut self,
+        paths: Vec<SafeStorePath<String>>,
+        substitute: bool,
+    ) -> Result<Vec<SafeStorePath<String>>, NixDaemonClientError>
+    where
+        R: AsyncReadExt + std::fmt::Debug + Unpin + Send,
+        W: AsyncWriteExt + std::fmt::Debug + Unpin + Send,
+    {
+        self.write_value(&Operation::QueryValidPaths).await?;
+        self.write_value(&paths).await?;
+
+        // write `substitute` bool
+        // https://snix.dev/docs/reference/nix-daemon-protocol/operations/#if-protocol-version-is-127-or-newer
+        if self.writer.version() >= ProtocolVersion::from_parts(1, 27) {
+            self.write_value(&substitute).await?;
+        }
+
+        self.writer
+            .flush()
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)?;
+
+        self.drain_stderr().await?;
+
+        #[allow(clippy::disallowed_types)]
+        let valid_paths: Vec<nix_compat::store_path::StorePath<String>> = self.read_value().await?;
+
+        return Ok(valid_paths.into_iter().map(SafeStorePath).collect());
+    }
+
+    #[instrument(skip(self))]
+    #[allow(clippy::disallowed_types)]
+    pub async fn collect_complete_closure(
+        &mut self,
+        root_path: &SafeStorePath<String>,
+    ) -> Result<Vec<SafeStorePath<String>>, NixDaemonClientError>
+    where
+        R: AsyncReadExt + std::fmt::Debug + Unpin + Send,
+        W: AsyncWriteExt + std::fmt::Debug + Unpin + Send,
+    {
+        let mut graph = HashMap::new();
+        let mut queue = vec![root_path.clone()];
+
+        while let Some(current_path) = queue.pop() {
+            if graph.contains_key(&current_path) {
+                continue;
+            }
+
+            // debug!(path = ?current_path, "querying path");
+
+            let path_info = self.query(&current_path).await?;
+
+            let Some(path_info) = path_info else {
+                return Err(NixDaemonClientError::NixDaemonOperationFailed(format!(
+                    "{current_path:?} does not exist in store"
+                )));
+            };
+
+            graph.insert(
+                current_path.clone(),
+                path_info
+                    .references
+                    .clone()
+                    .into_iter()
+                    .map(SafeStorePath)
+                    .collect(),
+            );
+
+            for reference in path_info.references {
+                let reference = SafeStorePath(reference);
+
+                if !graph.contains_key(&reference) {
+                    queue.push(reference);
+                }
+            }
+        }
+
+        let mut visited = HashSet::new();
+        let mut ordered = Vec::new();
+
+        // https://en.wikipedia.org/wiki/Topological_sorting#Depth-first_search
+        fn visit(
+            path: &SafeStorePath<String>,
+            graph: &HashMap<SafeStorePath<String>, Vec<SafeStorePath<String>>>,
+            visited: &mut HashSet<SafeStorePath<String>>,
+            ordered: &mut Vec<SafeStorePath<String>>,
+        ) {
+            if visited.contains(path) {
+                return;
+            }
+
+            visited.insert(path.clone());
+
+            if let Some(refs) = graph.get(path) {
+                for reference in refs {
+                    visit(reference, graph, visited, ordered);
+                }
+            }
+
+            ordered.push(path.clone());
+        }
+
+        visit(root_path, &graph, &mut visited, &mut ordered);
+
+        Ok(ordered)
+    }
+
+    // https://snix.dev/docs/reference/nix-daemon-protocol/operations/#querypathinfo
+    #[instrument(skip(self))]
+    #[allow(clippy::disallowed_types)]
+    pub async fn query(
+        &mut self,
+        path: &SafeStorePath<String>,
+    ) -> Result<Option<UnkeyedValidPathInfo>, NixDaemonClientError>
+    where
+        R: AsyncReadExt + std::fmt::Debug + Unpin + Send,
+        W: AsyncWriteExt + std::fmt::Debug + Unpin + Send,
+    {
+        if self.writer.version() < ProtocolVersion::from_parts(1, 17) {
+            return Err(NixDaemonClientError::NixDaemonProtocolVersion {
+                wanted: ProtocolVersion::from_parts(1, 17),
+                have: self.writer.version(),
+                operation: "QueryPathInfo".into(),
+            });
+        }
+
+        self.write_value(&Operation::QueryPathInfo).await?;
+        self.write_value(path).await?;
+        self.writer
+            .flush()
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)?;
+
+        self.drain_stderr().await?;
+
+        let success: bool = self.read_value().await?;
+
+        if !success {
+            return Ok(None);
+        }
+
+        // #[allow(clippy::disallowed_types)]
+        // let deriver: String = self
+        //     .reader
+        //     .read_value()
+        //     .await
+        //     .map_err(NixDaemonClientError::NixDaemonIO)?;
+        //
+        // info!("{deriver:?}");
+
+        #[allow(clippy::disallowed_types)]
+        let deriver: Option<nix_compat::store_path::StorePath<String>> = self.read_value().await?;
+
+        let nar_hash: String = self.read_value().await?;
+
+        #[allow(clippy::disallowed_types)]
+        let references: Vec<nix_compat::store_path::StorePath<String>> = self.read_value().await?;
+
+        let registration_time: u64 = self.read_value().await?;
+
+        let nar_size: u64 = self.read_value().await?;
+
+        let (ultimate, signatures, ca) =
+            if self.writer.version() >= ProtocolVersion::from_parts(1, 16) {
+                let ultimate: bool = self.read_value().await?;
+
+                let signatures: Vec<Signature<String>> = self.read_value().await?;
+
+                let ca: String = self.read_value().await?;
+
+                (
+                    ultimate,
+                    signatures,
+                    if ca.is_empty() {
+                        None
+                    } else {
+                        CAHash::from_nix_hex_str(&ca)
+                    },
+                )
+            } else {
+                (false, Vec::new(), None)
+            };
+
+        return Ok(Some(UnkeyedValidPathInfo {
+            deriver,
+            nar_hash,
+            references,
+            registration_time,
+            nar_size,
+            ultimate,
+            signatures,
+            ca,
+        }));
+    }
+
+    #[instrument(skip(self))]
+    #[allow(clippy::disallowed_types)]
+    pub async fn get_nar_stream(
+        &mut self,
+        path: &SafeStorePath<String>,
+        nar_size: u64,
+    ) -> Result<Take<&mut NixReader<R>>, NixDaemonClientError>
+    where
+        R: AsyncReadExt + std::fmt::Debug + Unpin + Send,
+        W: AsyncWriteExt + std::fmt::Debug + Unpin + Send,
+    {
+        if self.writer.version() < ProtocolVersion::from_parts(1, 17) {
+            return Err(NixDaemonClientError::NixDaemonProtocolVersion {
+                wanted: ProtocolVersion::from_parts(1, 17),
+                have: self.writer.version(),
+                operation: "NarFromPath".into(),
+            });
+        }
+
+        self.write_value(&Operation::NarFromPath).await?;
+        self.write_value(&path).await?;
+
+        self.writer
+            .flush()
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)?;
+        self.drain_stderr().await?;
+
+        Ok((&mut self.reader).take(nar_size))
+    }
+
+    #[instrument(skip(self, stream))]
+    #[allow(clippy::disallowed_types)]
+    pub async fn framed_write<RS>(&mut self, mut stream: RS) -> Result<(), NixDaemonClientError>
+    where
+        R: AsyncReadExt + std::fmt::Debug + Unpin + Send,
+        W: AsyncWriteExt + std::fmt::Debug + Unpin + Send,
+        RS: AsyncReadExt + Unpin + Send,
+    {
+        // 64KB buffer
+        let mut buffer = vec![0u8; 64 * 1024];
+
+        loop {
+            let bytes_read = stream
+                .read(&mut buffer)
+                .await
+                .map_err(NixDaemonClientError::NixDaemonIO)?;
+
+            if bytes_read == 0 {
+                // write a zero length to indicate the end of the framed stream
+                self.write_value(&0u64).await?;
+
+                break;
+            }
+
+            // write the chunk size
+            self.write_value(&(bytes_read as u64)).await?;
+
+            self.writer
+                .write_all(&buffer[..bytes_read])
+                .await
+                .map_err(NixDaemonClientError::NixDaemonIO)?;
+        }
+
+        self.writer
+            .flush()
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)?;
+
+        Ok(())
+    }
+
+    #[instrument(skip(self))]
+    pub async fn query_derivation_output_map(
+        &mut self,
+        store_path: &SafeStorePath<String>,
+    ) -> Result<BTreeMap<String, Option<SafeStorePath<String>>>, NixDaemonClientError>
+    where
+        R: AsyncReadExt + std::fmt::Debug + Unpin + Send,
+        W: AsyncWriteExt + std::fmt::Debug + Unpin + Send,
+    {
+        if self.writer.version() < ProtocolVersion::from_parts(1, 22) {
+            return Err(NixDaemonClientError::NixDaemonProtocolVersion {
+                wanted: ProtocolVersion::from_parts(1, 19),
+                have: self.writer.version(),
+                operation: "QueryMissing".into(),
+            });
+        }
+
+        self.write_value(&Operation::QueryDerivationOutputMap)
+            .await?;
+        self.write_value(&store_path).await?;
+        self.writer
+            .flush()
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)?;
+
+        self.drain_stderr().await?;
+
+        #[allow(clippy::disallowed_types)]
+        let map: BTreeMap<String, Option<nix_compat::store_path::StorePath<String>>> =
+            self.read_value().await?;
+
+        return Ok(map
+            .into_iter()
+            .map(|(key, value)| (key, value.map(SafeStorePath)))
+            .collect());
+    }
+
+    #[instrument(skip(self))]
+    #[allow(clippy::disallowed_types)]
+    pub async fn query_missing<S>(
+        &mut self,
+        derived_path: &Vec<DerivedPath<'_, S>>,
+    ) -> Result<QueryMissingResult, NixDaemonClientError>
+    where
+        R: AsyncReadExt + std::fmt::Debug + Unpin + Send,
+        W: AsyncWriteExt + std::fmt::Debug + Unpin + Send,
+        S: Borrow<str> + std::fmt::Debug + Sync,
+    {
+        if self.writer.version() < ProtocolVersion::from_parts(1, 19) {
+            return Err(NixDaemonClientError::NixDaemonProtocolVersion {
+                wanted: ProtocolVersion::from_parts(1, 19),
+                have: self.writer.version(),
+                operation: "QueryMissing".into(),
+            });
+        }
+
+        self.write_value(&Operation::QueryMissing).await?;
+        self.write_value(derived_path).await?;
+        self.writer
+            .flush()
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)?;
+
+        self.drain_stderr().await?;
+
+        self.read_value().await
+    }
+
+    #[instrument(skip(self))]
+    #[allow(clippy::disallowed_types)]
+    pub async fn build<S>(
+        &mut self,
+        derived_paths: &Vec<DerivedPath<'_, S>>,
+    ) -> Result<(), NixDaemonClientError>
+    where
+        R: AsyncReadExt + std::fmt::Debug + Unpin + Send,
+        W: AsyncWriteExt + std::fmt::Debug + Unpin + Send,
+        S: Borrow<str> + std::fmt::Debug + Sync,
+    {
+        if self.writer.version() < ProtocolVersion::from_parts(1, 19) {
+            info!(
+                "daemon does not support QueryMissing, cannot log which paths will be built or substituted"
+            );
+        } else {
+            let missing = self.query_missing(derived_paths).await?;
+
+            trace!("missing: {missing:?}");
+
+            if !missing.will_build.is_empty() {
+                info!(
+                    "The following {} paths will be built: \n\t{}",
+                    missing
+                        .will_build
+                        .len()
+                        .if_supports_color(Stream::Stderr, |x| x.green()),
+                    missing
+                        .will_build
+                        .into_iter()
+                        .map(|x| x.to_absolute_path())
+                        .join("\n\t")
+                );
+            }
+
+            if !missing.will_substitute.is_empty() {
+                info!(
+                    "{} paths will be substituted for {} bytes: \n\t{}",
+                    missing
+                        .will_substitute
+                        .len()
+                        .if_supports_color(Stream::Stderr, |x| x.green()),
+                    missing.download_size,
+                    missing
+                        .will_substitute
+                        .into_iter()
+                        .map(|x| x.to_absolute_path())
+                        .join("\n\t"),
+                );
+            }
+        }
+
+        self.write_value(&Operation::BuildPaths).await?;
+        self.write_value(derived_paths).await?;
+
+        if self.writer.version() >= ProtocolVersion::from_parts(1, 15) {
+            self.write_value(&0u64).await?;
+        }
+
+        self.writer
+            .flush()
+            .await
+            .map_err(NixDaemonClientError::NixDaemonIO)?;
+
+        self.drain_stderr().await?;
+
+        let value = self.read_value::<u64>().await?;
+
+        trace!("end of build: read u64 {value:?}");
+
+        Ok(())
+    }
+
+    #[instrument(skip(self, nar_stream))]
+    #[allow(clippy::disallowed_types)]
+    pub async fn add_to_store_nar<RS>(
+        &mut self,
+        data: WireAddToStoreNarRequest,
+        nar_stream: RS,
+    ) -> Result<(), NixDaemonClientError>
+    where
+        R: AsyncReadExt + std::fmt::Debug + Unpin + Send,
+        W: AsyncWriteExt + std::fmt::Debug + Unpin + Send,
+        RS: AsyncReadExt + Unpin + Send,
+    {
+        // non-framed data writes are not implemented
+        if self.writer.version() < ProtocolVersion::from_parts(1, 23) {
+            return Err(NixDaemonClientError::NixDaemonProtocolVersion {
+                wanted: ProtocolVersion::from_parts(1, 23),
+                have: self.writer.version(),
+                operation: "AddToStoreNar".into(),
+            });
+        }
+
+        self.write_value(&Operation::AddToStoreNar).await?;
+        self.write_value(&data).await?;
+
+        self.framed_write(nar_stream).await?;
+        self.drain_stderr().await?;
+
+        Ok(())
+    }
+}
+
+impl<S: Borrow<str> + std::fmt::Debug + Sync> NixSerialize for DerivedPath<'_, S> {
+    async fn serialize<W>(&self, writer: &mut W) -> Result<(), W::Error>
+    where
+        W: NixWrite,
+    {
+        writer.write_value(&self).await
+    }
+}
+
+// https://snix.dev/docs/reference/nix-daemon-protocol/types/#derivedpath
+impl<S: Borrow<str> + std::fmt::Debug + Sync> NixSerialize for &DerivedPath<'_, S> {
+    async fn serialize<W>(&self, writer: &mut W) -> Result<(), W::Error>
+    where
+        W: NixWrite,
+    {
+        let output_names = match self.outputs {
+            DerivedPathOutput::OutputNames(outputs) => outputs.join(","),
+        };
+
+        let output_spec = format!("{}!{}", self.store_path.to_absolute_path(), output_names);
+
+        writer.write_value(&output_spec).await
+    }
+}
+
+impl NixSerialize for WireAddToStoreNarRequest {
+    async fn serialize<W>(&self, writer: &mut W) -> Result<(), W::Error>
+    where
+        W: NixWrite,
+    {
+        writer.write_value(&self.path).await?;
+        #[allow(clippy::disallowed_types)]
+        writer
+            .write_value(
+                &self
+                    .deriver
+                    .clone()
+                    .map(Into::<nix_compat::store_path::StorePath<String>>::into),
+            )
+            .await?;
+        writer.write_value(&self.nar_hash).await?;
+        writer.write_value(&self.references).await?;
+        writer.write_value(&self.registration_time).await?;
+        writer.write_value(&self.nar_size).await?;
+        writer.write_value(&self.ultimate).await?;
+        writer.write_value(&self.signatures).await?;
+        writer.write_value(&self.ca).await?;
+        writer.write_value(&self.repair).await?;
+        writer.write_value(&self.dont_check_sigs).await
+    }
+}
+
+impl NixDeserialize for QueryMissingResult {
+    async fn try_deserialize<R>(reader: &mut R) -> Result<Option<Self>, R::Error>
+    where
+        R: ?Sized + NixRead + Send,
+    {
+        let will_build: Option<Vec<SafeStorePath<String>>> = reader.try_read_value().await?;
+        let will_substitute: Option<Vec<SafeStorePath<String>>> = reader.try_read_value().await?;
+        let unknown: Option<Vec<SafeStorePath<String>>> = reader.try_read_value().await?;
+        let download_size = reader.try_read_number().await?;
+        let nar_size = reader.try_read_number().await?;
+
+        if let Some(will_build) = will_build
+            && let Some(will_substitute) = will_substitute
+            && let Some(unknown) = unknown
+            && let Some(download_size) = download_size
+            && let Some(nar_size) = nar_size
+        {
+            Ok(Some(QueryMissingResult {
+                will_build: will_build.into_iter().collect(),
+                will_substitute: will_substitute.into_iter().collect(),
+                _unknown: unknown.into_iter().collect(),
+                download_size,
+                _nar_size: nar_size,
+            }))
+        } else {
+            Ok(None)
+        }
+    }
+}
diff --git a/crates/nix_client/src/store_path.rs b/crates/nix_client/src/store_path.rs
new file mode 100644
index 0000000..0347a72
--- /dev/null
+++ b/crates/nix_client/src/store_path.rs
@@ -0,0 +1,177 @@
+use miette::Diagnostic;
+use nix_compat::wire::ser::NixSerialize;
+use serde::Deserialize;
+use std::hash::Hash;
+use thiserror::Error;
+
+#[derive(Debug, Diagnostic, Error)]
+#[diagnostic(code(wire::SnixStorePath))]
+#[error("Failed to parse store path {path:?}")]
+pub struct StorePathError {
+    path: String,
+    #[source]
+    error: nix_compat::store_path::Error,
+}
+
+/// This type exists to restrict `StorePath` usage to only methods that deal with
+/// absolute paths. By default, the `StorePath` type implements Display that
+/// does not include `/nix/store/` can introduce many hard to catch bugs.
+///
+///
+/// If <https://github.com/rust-lang/rust-clippy/issues/8581>
+/// is ever closed, this can be dropped from the codebase.
+#[derive(Clone)]
+#[allow(clippy::disallowed_types)]
+pub struct SafeStorePath<S>(pub nix_compat::store_path::StorePath<S>);
+
+#[allow(clippy::disallowed_types)]
+impl<S> SafeStorePath<S> {
+    pub fn from_absolute_path<'a>(s: &'a [u8]) -> Result<SafeStorePath<S>, StorePathError>
+    where
+        S: From<&'a str> + AsRef<str>,
+    {
+        Ok(Self(
+            nix_compat::store_path::StorePath::from_absolute_path(s).map_err(|error| {
+                StorePathError {
+                    path: String::from_utf8_lossy(s).to_string(),
+                    error,
+                }
+            })?,
+        ))
+    }
+
+    pub fn into_inner(self) -> nix_compat::store_path::StorePath<S> {
+        self.0
+    }
+
+    pub fn from_name_and_digest<'a>(name: &'a str, digest: &[u8]) -> Result<Self, StorePathError>
+    where
+        S: From<&'a str> + AsRef<str>,
+    {
+        Ok(Self(
+            nix_compat::store_path::StorePath::from_name_and_digest(name, digest).map_err(
+                |error| StorePathError {
+                    path: format!("raw name & digest: {digest:?}-{name:?}"),
+                    error,
+                },
+            )?,
+        ))
+    }
+
+    pub fn to_absolute_path(&self) -> String
+    where
+        S: AsRef<str>,
+    {
+        self.0.to_absolute_path()
+    }
+
+    pub fn digest(&self) -> &[u8; nix_compat::store_path::DIGEST_SIZE]
+    where
+        S: AsRef<str>,
+    {
+        self.0.digest()
+    }
+
+    pub fn name(&self) -> &S
+    where
+        S: AsRef<str>,
+    {
+        self.0.name()
+    }
+}
+
+#[allow(clippy::disallowed_types)]
+impl<'de, S> Deserialize<'de> for SafeStorePath<S>
+where
+    nix_compat::store_path::StorePath<S>: Deserialize<'de>,
+    S: AsRef<str>,
+{
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        Ok(SafeStorePath(
+            nix_compat::store_path::StorePath::deserialize(deserializer)?,
+        ))
+    }
+}
+
+impl<S> PartialEq for SafeStorePath<S>
+where
+    S: AsRef<str>,
+{
+    fn eq(&self, other: &Self) -> bool {
+        self.0.eq(&other.0)
+    }
+}
+
+impl<S> Hash for SafeStorePath<S>
+where
+    S: AsRef<str>,
+{
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.0.hash(state);
+    }
+}
+
+impl<S> Eq for SafeStorePath<S> where S: AsRef<str> {}
+
+impl<S> NixSerialize for SafeStorePath<S>
+where
+    S: AsRef<str>,
+{
+    fn serialize<W>(&self, writer: &mut W) -> impl Future<Output = Result<(), W::Error>> + Send
+    where
+        W: nix_compat::wire::ser::NixWrite,
+    {
+        self.0.serialize(writer)
+    }
+}
+
+impl<S> NixSerialize for &SafeStorePath<S>
+where
+    S: AsRef<str>,
+{
+    fn serialize<W>(&self, writer: &mut W) -> impl Future<Output = Result<(), W::Error>> + Send
+    where
+        W: nix_compat::wire::ser::NixWrite,
+    {
+        self.0.serialize(writer)
+    }
+}
+
+impl nix_compat::wire::de::NixDeserialize for SafeStorePath<String> {
+    async fn try_deserialize<R>(reader: &mut R) -> Result<Option<Self>, R::Error>
+    where
+        R: ?Sized + nix_compat::wire::de::NixRead + Send,
+    {
+        if let Some(store_path) = reader.try_read_value().await? {
+            Ok(Some(SafeStorePath(store_path)))
+        } else {
+            Ok(None)
+        }
+    }
+}
+
+#[allow(clippy::disallowed_types)]
+impl<S> From<nix_compat::store_path::StorePath<S>> for SafeStorePath<S> {
+    fn from(value: nix_compat::store_path::StorePath<S>) -> Self {
+        SafeStorePath(value)
+    }
+}
+
+#[allow(clippy::disallowed_types)]
+impl<S> From<SafeStorePath<S>> for nix_compat::store_path::StorePath<S> {
+    fn from(value: SafeStorePath<S>) -> nix_compat::store_path::StorePath<S> {
+        value.into_inner()
+    }
+}
+
+impl<S> std::fmt::Debug for SafeStorePath<S>
+where
+    S: AsRef<str>,
+{
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(&self.0.to_absolute_path())
+    }
+}