git.delta.rocks / fleet / refs/commits / 7bc4be6bcef6

difftreelog

feat remowt integration

kyumtlkpYaroslav Bolyukin2026-06-05parent: #7bbada6.patch.diff
in: push-kyumtlkprzyo

46 files changed

modifiedCargo.lockdiffbeforeafterboth
after · Cargo.lock
576 packageslockfile v4
modifiedCargo.tomldiffbeforeafterboth
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -6,22 +6,28 @@
 package.rust-version = "1.89.0"
 
 [workspace.dependencies]
-better-command = { path = "./crates/better-command" }
 fleet-base = { path = "./crates/fleet-base" }
 fleet-shared = { path = "./crates/fleet-shared" }
 nix-eval = { path = "./crates/nix-eval" }
 nixlike = { path = "./crates/nixlike" }
 opentelemetry-exporter-env = { path = "./crates/opentelemetry-exporter-env" }
+remowt-fleet = { path = "./crates/remowt-fleet" }
 
-abort-on-drop = "0.2"
+remowt-client = "0.1.3"
+remowt-endpoints = "0.1.3"
+remowt-link-shared = "0.1.3"
+remowt-plugin = "0.1.3"
+remowt-ui-prompt = "0.1.3"
+
+bifrostlink = "0.2.3"
+
+uuid = { version = "1", features = ["v4"] }
+
 age = { version = "0.11", features = ["plugin", "ssh"] }
-age-core = "0.11"
 anyhow = "1.0"
-async-trait = "0.1.88"
-axum = { version = "0.8.7", features = ["http2", "macros", "ws"] }
-axum-extra = { version = "0.12.2", features = ["typed-header"] }
 base64 = "0.22.1"
 bindgen = "0.72.0"
+camino = "1.2.2"
 chrono = { version = "0.4.41", features = ["serde"] }
 clap = { version = "4.5", features = ["derive", "env", "unicode", "wrap_help"] }
 clap_complete = "4.5"
@@ -29,19 +35,16 @@
 cxx-build = "1.0.168"
 ed25519-dalek = "2.1"
 futures = "0.3.31"
-futures-util = { version = "0.3.31", features = ["sink"] }
 hex = "0.4.3"
 hmac = "0.12"
 hostname = "0.4.1"
 human-repr = "1.1"
-hyper = "1.8.1"
 indicatif = "0.18"
 indoc = "2.0.6"
 itertools = "0.14.0"
 linked-hash-map = "0.5.6"
 nix = { version = "0.31.2", features = ["fs", "user"] }
 nom = "8.0.0"
-openssh = "0.11.5"
 opentelemetry = "0.31.0"
 opentelemetry-appender-tracing = "0.31.1"
 opentelemetry-otlp = { version = "0.31.0", features = ["grpc-tonic", "gzip-tonic", "http-json", "reqwest-rustls"] }
@@ -50,21 +53,16 @@
 peg = "0.8.5"
 pkg-config = "0.3.30"
 rand = "0.10.0"
-regex = "1.11"
 serde = { version = "1.0", features = ["derive"] }
-serde-transcode = "1.1.1"
 serde_json = "1.0"
 sha2 = "0.10"
 shlex = "1.3"
 tabled = "0.20.0"
 tempfile = "3.20"
 test-log = { version = "0.2.19", features = ["trace"] }
-tf-provider = "0.2.2"
 thiserror = "2.0.12"
 time = "0.3.41"
 tokio = { version = "1.45.1", features = ["fs", "macros", "rt", "rt-multi-thread", "sync", "time"] }
-tokio-util = "0.7.17"
-toml_edit = "0.23.10"
 tracing = "0.1"
 tracing-indicatif = "0.3.13"
 tracing-opentelemetry = "0.32.1"
modifiedcmds/fleet/Cargo.tomldiffbeforeafterboth
--- a/cmds/fleet/Cargo.toml
+++ b/cmds/fleet/Cargo.toml
@@ -8,15 +8,13 @@
 default-run = "fleet"
 
 [dependencies]
-age = { workspace = true, features = ["armor"] }
 anyhow.workspace = true
-better-command.workspace = true
+camino.workspace = true
 clap.workspace = true
 clap_complete.workspace = true
 fleet-base.workspace = true
-fleet-shared.workspace = true
 nix-eval.workspace = true
-nixlike.workspace = true
+remowt-fleet.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 tempfile.workspace = true
@@ -24,30 +22,17 @@
 tracing.workspace = true
 tracing-subscriber.workspace = true
 
-abort-on-drop.workspace = true
-age-core.workspace = true
-async-trait.workspace = true
-base64.workspace = true
-chrono.workspace = true
 futures.workspace = true
-hostname.workspace = true
 itertools.workspace = true
-openssh.workspace = true
-peg.workspace = true
-regex.workspace = true
 shlex.workspace = true
 tabled.workspace = true
-time = { workspace = true, features = ["serde"] }
-tokio-util = { workspace = true, features = ["codec"] }
 
 human-repr = { workspace = true, optional = true }
 indicatif = { workspace = true, optional = true }
-nom.workspace = true
 opentelemetry.workspace = true
 opentelemetry-appender-tracing.workspace = true
 opentelemetry-exporter-env.workspace = true
 opentelemetry_sdk.workspace = true
-thiserror.workspace = true
 tracing-indicatif = { workspace = true, optional = true }
 tracing-opentelemetry.workspace = true
 
@@ -58,6 +43,5 @@
 	"dep:tracing-indicatif",
 	"dep:indicatif",
 	"dep:human-repr",
-	"better-command/indicatif",
 	"nix-eval/indicatif",
 ]
modifiedcmds/fleet/src/cmds/build_systems.rsdiffbeforeafterboth
--- a/cmds/fleet/src/cmds/build_systems.rs
+++ b/cmds/fleet/src/cmds/build_systems.rs
@@ -1,6 +1,7 @@
 use std::{env::current_dir, os::unix::fs::symlink, path::PathBuf};
 
 use anyhow::Result;
+use camino::Utf8PathBuf;
 use clap::Parser;
 use fleet_base::{
 	deploy::{DeployAction, deploy_task, upload_task},
@@ -29,7 +30,7 @@
 	build_attr: String,
 }
 
-async fn build_task(config: Config, hostname: String, build_attr: &str) -> Result<PathBuf> {
+async fn build_task(config: Config, hostname: String, build_attr: &str) -> Result<Utf8PathBuf> {
 	info!("building");
 	let host = config.host(&hostname)?;
 	// let action = Action::from(self.subcommand.clone());
@@ -42,17 +43,20 @@
 	// We already have system profiles for backups.
 	if !host.local {
 		info!("adding gc root");
-		let mut cmd = config.local_host().cmd("nix").await?;
-		cmd.arg("build")
-			.comparg(
-				"--profile",
-				format!(
-					"/nix/var/nix/profiles/{}-{hostname}",
-					config.data.gc_root_prefix
-				),
-			)
-			.arg(&out_output);
-		cmd.sudo().run_nix().await?;
+		let local = config.local_host();
+		let plugin_id = local.ensure_nix_plugin().await?;
+		let nix = local
+			.remowt()
+			.await?
+			.plugin_endpoints::<remowt_fleet::NixClient<_>>(plugin_id);
+		let profile = format!(
+			"/nix/var/nix/profiles/{}-{hostname}",
+			config.data.gc_root_prefix
+		);
+		nix.switch_profile(profile, out_output.clone())
+			.await
+			.map_err(|e| anyhow::anyhow!("{e:?}"))?
+			.map_err(|e| anyhow::anyhow!("{e}"))?;
 	}
 
 	Ok(out_output)
modifiedcmds/fleet/src/cmds/tf.rsdiffbeforeafterboth
--- a/cmds/fleet/src/cmds/tf.rs
+++ b/cmds/fleet/src/cmds/tf.rs
@@ -1,6 +1,7 @@
-use std::{collections::BTreeMap, ffi::OsString, path::PathBuf};
+use std::{collections::BTreeMap, ffi::OsString};
 
 use anyhow::{Context, Result};
+use camino::Utf8PathBuf;
 use clap::Parser;
 use fleet_base::host::Config;
 use nix_eval::nix_go;
@@ -40,7 +41,7 @@
 			let system = &config.local_system;
 			let config = &config.flake_outputs;
 			let data = nix_go!(config.tf({ system }));
-			let data: PathBuf = spawn_blocking(move || data.build("out"))
+			let data: Utf8PathBuf = spawn_blocking(move || data.build("out"))
 				.await
 				.expect("tf.json derivation should not fail")?;
 			let data = fs::read(&data).await?;
modifiedcmds/fleet/src/main.rsdiffbeforeafterboth
--- a/cmds/fleet/src/main.rs
+++ b/cmds/fleet/src/main.rs
@@ -4,9 +4,10 @@
 // pub(crate) mod command;
 pub(crate) mod extra_args;
 
-use std::{env, ffi::OsString, process::ExitCode, sync::Arc};
+use std::{process::ExitCode, sync::Arc};
 
-use anyhow::{Result, bail};
+use anyhow::{Context as _, Result, bail};
+use camino::Utf8PathBuf;
 use clap::{CommandFactory, Parser};
 use cmds::{
 	build_systems::{BuildSystems, Deploy},
@@ -23,7 +24,8 @@
 #[cfg(feature = "indicatif")]
 use indicatif::{ProgressState, ProgressStyle};
 use nix_eval::{
-	gc_register_my_thread, gc_unregister_my_thread, init_libraries, init_tokio_for_nix,
+	add_file_to_store, gc_register_my_thread, gc_unregister_my_thread, init_libraries,
+	init_tokio_for_nix,
 };
 use opentelemetry::trace::TracerProvider;
 use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
@@ -48,24 +50,21 @@
 		}
 		let tasks = FuturesUnordered::new();
 		for entry in std::fs::read_dir(&prefetch_dir)? {
-			tasks.push(async {
-				let entry = entry?;
-				if !entry.metadata()?.is_file() {
-					bail!("only files should exist in prefetch directory");
-				}
-				let span = info_span!(
-					"prefetching",
-					name = entry.file_name().to_string_lossy().as_ref()
-				);
-				let mut path = OsString::new();
-				path.push("file://");
-				path.push(entry.path());
-
-				let mut status = config.local_host().cmd("nix").await?;
-				status.args(&config.nix_args);
-				status.arg("store").arg("prefetch-file").arg(path);
-				status.run_nix_string().instrument(span).await?;
-				Ok(())
+			let entry = entry?;
+			if !entry.metadata()?.is_file() {
+				bail!("only files should exist in prefetch directory");
+			}
+			let name = entry.file_name().to_string_lossy().into_owned();
+			let path =
+				Utf8PathBuf::try_from(entry.path()).context("prefetch path should be utf8")?;
+			let span = info_span!("prefetching", name = %name);
+			tasks.push(async move {
+				let added = tokio::task::spawn_blocking(move || add_file_to_store(&name, &path))
+					.instrument(span.clone())
+					.await??;
+				let _g = span.enter();
+				info!("{} -> {}", added.hash, added.store_path);
+				anyhow::Ok(())
 			});
 		}
 		tasks.try_collect::<Vec<()>>().await?;
@@ -179,9 +178,7 @@
 	let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
 
 	let reg = tracing_subscriber::registry().with({
-		let sub = tracing_subscriber::fmt::layer()
-			.without_time()
-			.with_target(false);
+		let sub = tracing_subscriber::fmt::layer().without_time();
 		#[cfg(feature = "indicatif")]
 		let sub = sub.with_writer(indicatif_layer.get_stderr_writer());
 		sub.with_filter(filter) // .without,
@@ -262,14 +259,10 @@
 }
 
 async fn main_real(opts: RootOpts) -> Result<()> {
-	let nix_args = std::env::var_os("NIX_ARGS")
-		.map(|a| extra_args::parse_os(&a))
-		.transpose()?
-		.unwrap_or_default();
-	let config = opts.fleet_opts.build(
-		nix_args,
-		matches!(opts.command, Opts::Deploy(_) | Opts::BuildSystems(_)),
-	)?;
+	let config = opts.fleet_opts.build(matches!(
+		opts.command,
+		Opts::Deploy(_) | Opts::BuildSystems(_)
+	))?;
 
 	match run_command(&config, opts.fleet_opts, opts.command).await {
 		Ok(()) => {
modifiedcmds/install-secrets/src/main.rsdiffbeforeafterboth
--- a/cmds/install-secrets/src/main.rs
+++ b/cmds/install-secrets/src/main.rs
@@ -227,7 +227,6 @@
 				.from_env_lossy(),
 		)
 		.without_time()
-		.with_target(false)
 		.init();
 
 	let opts = Opts::parse();
addedcmds/nix-plugin-fleet/Cargo.tomldiffbeforeafterboth
--- /dev/null
+++ b/cmds/nix-plugin-fleet/Cargo.toml
@@ -0,0 +1,11 @@
+[package]
+name = "nix-plugin-fleet"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+
+[lib]
+crate-type = ["cdylib"]
+
+[dependencies]
+fleet-base.workspace = true
addedcmds/nix-plugin-fleet/src/lib.rsdiffbeforeafterboth
--- /dev/null
+++ b/cmds/nix-plugin-fleet/src/lib.rs
@@ -0,0 +1,7 @@
+use fleet_base::primops::init_primops;
+
+/// SAFETY: expected plugin dynamic library entry point
+#[unsafe(no_mangle)]
+fn nix_plugin_entry() {
+	init_primops();
+}
addedcmds/remowt-plugin-fleet/Cargo.tomldiffbeforeafterboth
--- /dev/null
+++ b/cmds/remowt-plugin-fleet/Cargo.toml
@@ -0,0 +1,11 @@
+[package]
+name = "remowt-plugin-fleet"
+description = "Remowt plugin exposing fleet's Nix endpoint to a running agent"
+version.workspace = true
+edition = "2021"
+
+[dependencies]
+remowt-fleet.workspace = true
+remowt-plugin.workspace = true
+
+anyhow.workspace = true
addedcmds/remowt-plugin-fleet/src/main.rsdiffbeforeafterboth
--- /dev/null
+++ b/cmds/remowt-plugin-fleet/src/main.rs
@@ -0,0 +1,8 @@
+use remowt_fleet::{Nix, init_libraries};
+
+fn main() -> anyhow::Result<()> {
+	remowt_plugin::run(|rpc| {
+		init_libraries();
+		Nix.register_endpoints(rpc);
+	})
+}
deletedcmds/repl-plugin-unstable/Cargo.tomldiffbeforeafterboth
--- a/cmds/repl-plugin-unstable/Cargo.toml
+++ /dev/null
@@ -1,11 +0,0 @@
-[package]
-name = "repl-plugin-unstable"
-version.workspace = true
-edition.workspace = true
-rust-version.workspace = true
-
-[lib]
-crate-type = ["cdylib"]
-
-[dependencies]
-fleet-base.workspace = true
deletedcmds/repl-plugin-unstable/src/lib.rsdiffbeforeafterboth
--- a/cmds/repl-plugin-unstable/src/lib.rs
+++ /dev/null
@@ -1,7 +0,0 @@
-use fleet_base::primops::init_primops;
-
-/// SAFETY: expected plugin dynamic library entry point
-#[unsafe(no_mangle)]
-fn nix_plugin_entry() {
-	init_primops();
-}
deletedcmds/terraform-provider-fleet/Cargo.tomldiffbeforeafterboth
--- a/cmds/terraform-provider-fleet/Cargo.toml
+++ /dev/null
@@ -1,13 +0,0 @@
-[package]
-name = "terraform-provider-fleet"
-version.workspace = true
-edition.workspace = true
-rust-version.workspace = true
-
-[dependencies]
-anyhow.workspace = true
-serde = { workspace = true, features = ["derive"] }
-tokio.workspace = true
-
-async-trait.workspace = true
-tf-provider.workspace = true
deletedcmds/terraform-provider-fleet/src/main.rsdiffbeforeafterboth
--- a/cmds/terraform-provider-fleet/src/main.rs
+++ /dev/null
@@ -1,5 +0,0 @@
-fn main() {
-	panic!(
-		"this is a stub, real provider is in development, I just don't want to keep it in different branch."
-	)
-}
deletedcrates/better-command/Cargo.tomldiffbeforeafterboth
--- a/crates/better-command/Cargo.toml
+++ /dev/null
@@ -1,16 +0,0 @@
-[package]
-name = "better-command"
-version.workspace = true
-edition.workspace = true
-rust-version.workspace = true
-
-[dependencies]
-regex.workspace = true
-serde.workspace = true
-serde_json.workspace = true
-tracing.workspace = true
-
-tracing-indicatif = { workspace = true, optional = true }
-
-[features]
-indicatif = ["tracing-indicatif"]
deletedcrates/better-command/src/handler.rsdiffbeforeafterboth
--- a/crates/better-command/src/handler.rs
+++ /dev/null
@@ -1,335 +0,0 @@
-//! Collection of handlers, which transform program-specific stdout format to tracing
-
-use std::{
-	collections::HashMap,
-	sync::{Arc, LazyLock, Mutex},
-};
-
-use regex::Regex;
-use serde::Deserialize;
-use tracing::{Span, info, info_span, warn};
-#[cfg(feature = "indicatif")]
-use tracing_indicatif::span_ext::IndicatifSpanExt as _;
-
-pub trait Handler: Send {
-	fn handle_line(&mut self, e: &str);
-}
-
-/// Handler wrapper, which can be cloned.
-pub struct ClonableHandler<H>(Arc<Mutex<H>>);
-impl<H> Clone for ClonableHandler<H> {
-	fn clone(&self) -> Self {
-		Self(self.0.clone())
-	}
-}
-impl<H> ClonableHandler<H> {
-	pub fn new(inner: H) -> Self {
-		Self(Arc::new(Mutex::new(inner)))
-	}
-}
-impl<H: Handler> Handler for ClonableHandler<H> {
-	fn handle_line(&mut self, e: &str) {
-		self.0.lock().unwrap().handle_line(e)
-	}
-}
-
-/// Converts command output to tracing lines
-pub struct PlainHandler;
-impl Handler for PlainHandler {
-	fn handle_line(&mut self, e: &str) {
-		info!(target: "log", "{e}");
-	}
-}
-
-/// Ignores output
-pub struct NoopHandler;
-impl Handler for NoopHandler {
-	fn handle_line(&mut self, _e: &str) {}
-}
-
-/// Transform nix internal-json logs to tracing spans.
-#[derive(Default)]
-pub struct NixHandler {
-	spans: HashMap<u64, Span>,
-}
-#[derive(Deserialize, Debug)]
-#[serde(untagged)]
-enum LogField {
-	String(String),
-	Num(u64),
-}
-
-/// Nix internal-json log line type
-#[derive(Deserialize, Debug)]
-#[serde(rename_all = "camelCase", tag = "action")]
-#[allow(dead_code)]
-enum NixLog {
-	Msg {
-		level: u32,
-		msg: String,
-		raw_msg: Option<String>,
-	},
-	Start {
-		id: u64,
-		level: u32,
-		#[serde(default)]
-		fields: Vec<LogField>,
-		text: String,
-		#[serde(rename = "type")]
-		typ: u32,
-	},
-	Stop {
-		id: u64,
-	},
-	Result {
-		id: u64,
-		#[serde(rename = "type")]
-		typ: u32,
-		#[serde(default)]
-		fields: Vec<LogField>,
-	},
-}
-fn process_message(m: &str) -> String {
-	// Supposed to remove formatting characters except colors, as some programs try to reset cursor position etc.
-	static OSC_CLEANER: LazyLock<Regex> =
-		LazyLock::new(|| Regex::new(r"\x1B\]([^\x07\x1C]*[\x07\x1C])?|\r").unwrap());
-	static DETABBER: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"\t").unwrap());
-	let m = OSC_CLEANER.replace_all(m, "");
-	// Indicatif can't format tabs. This is not the correct tab formatting, as correct one should be aligned,
-	// and not just be replaced with the constant number of spaces, but it's ok for now, as statuses are single-line.
-	DETABBER.replace_all(m.as_ref(), "  ").to_string()
-}
-impl Handler for NixHandler {
-	fn handle_line(&mut self, e: &str) {
-		if let Some(e) = e.strip_prefix("@nix ") {
-			let log: NixLog = match serde_json::from_str(e) {
-				Ok(l) => l,
-				Err(err) => {
-					warn!("failed to parse nix log line {:?}: {}", e, err);
-					return;
-				}
-			};
-			match log {
-				NixLog::Msg { msg, raw_msg, .. } => {
-					#[allow(clippy::nonminimal_bool)]
-					if !(msg.starts_with("\u{1b}[35;1mwarning:\u{1b}[0m Git tree '")
-						&& msg.ends_with("' is dirty"))
-						&& !msg.starts_with(
-							"\u{1b}[35;1mwarning:\u{1b}[0m not writing modified lock file of flake",
-						) && msg
-						!= "\u{1b}[35;1mwarning:\u{1b}[0m \u{1b}[31;1merror:\u{1b}[0m SQLite database '\u{1b}[35;1m/nix/var/nix/db/db.sqlite\u{1b}[0m' is busy"
-					{
-						if let Some(raw_msg) = raw_msg {
-							if !msg.is_empty() {
-								info!(target: "nix", "{}\n{}", raw_msg.trim_end(), msg.trim_end())
-							} else {
-								info!(target: "nix", "{}", raw_msg.trim_end())
-							}
-						} else {
-							info!(target: "nix", "{}", msg.trim_end())
-						}
-					}
-				}
-				NixLog::Start {
-					ref fields,
-					typ,
-					id,
-					..
-				} if typ == 105 && !fields.is_empty() => {
-					if let [LogField::String(drv), ..] = &fields[..] {
-						let mut drv = drv.as_str();
-						if let Some(pkg) = drv.strip_prefix("/nix/store/") {
-							let mut it = pkg.splitn(2, '-');
-							it.next();
-							if let Some(pkg) = it.next() {
-								drv = pkg;
-							}
-						}
-						info!(target: "nix","building {}", drv);
-						let span = info_span!("build", drv);
-						#[cfg(feature = "indicatif")]
-						span.pb_start();
-						self.spans.insert(id, span);
-					} else {
-						warn!("bad build log: {:?}", log)
-					}
-				}
-				NixLog::Start {
-					ref fields,
-					typ,
-					id,
-					..
-				} if typ == 100 && fields.len() >= 3 => {
-					if let [
-						LogField::String(drv),
-						LogField::String(from),
-						LogField::String(to),
-						..,
-					] = &fields[..]
-					{
-						let mut drv = drv.as_str();
-
-						if let Some(pkg) = drv.strip_prefix("/nix/store/") {
-							let mut it = pkg.splitn(2, '-');
-							it.next();
-							if let Some(pkg) = it.next() {
-								drv = pkg;
-							}
-						}
-						info!(target: "nix","copying {} {} -> {}", drv, from, to);
-						let span = info_span!("copy", from, to, drv);
-						#[cfg(feature = "indicatif")]
-						span.pb_start();
-						self.spans.insert(id, span);
-					} else {
-						warn!("bad copy log: {:?}", log)
-					}
-				}
-				NixLog::Start { text, typ, id, .. }
-					if typ == 0 || typ == 102 || typ == 103 || typ == 104 =>
-				{
-					if !text.is_empty()
-						&& text != "querying info about missing paths"
-						&& text != "copying 0 paths"
-						// Too much spam on lazy-trees branch
-						&& !(text.starts_with("copying '") && text.ends_with("' to the store"))
-					{
-						let span = info_span!("job");
-						#[cfg(feature = "indicatif")]
-						{
-							span.pb_start();
-							span.pb_set_message(&process_message(text.trim()));
-						}
-						self.spans.insert(id, span);
-						info!(target: "nix", "{}", text);
-					}
-				}
-				NixLog::Start {
-					text,
-					level: 0,
-					typ: 108,
-					..
-				} if text.is_empty() => {
-					// Cache lookup? Coupled with copy log
-				}
-				NixLog::Start {
-					text,
-					level: 4,
-					typ: 109,
-					..
-				} if text.starts_with("querying info about ") => {
-					// Cache lookup
-				}
-				NixLog::Start {
-					text,
-					level: 4,
-					typ: 101,
-					..
-				} if text.starts_with("downloading ") => {
-					// NAR downloading, coupled with copy log
-				}
-				NixLog::Start {
-					text,
-					level: 1,
-					typ: 111,
-					..
-				} if text.starts_with("waiting for a machine to build ") => {
-					// Useless repeating notification about build
-				}
-				NixLog::Start {
-					text,
-					level: 3,
-					typ: 111,
-					..
-				} if text.starts_with("resolved derivation: ") => {
-					// CA resolved
-				}
-				NixLog::Start {
-					text,
-					level: 1,
-					typ: 111,
-					id,
-					..
-				} if text.starts_with("waiting for lock on ") => {
-					let mut drv = text.strip_prefix("waiting for lock on ").unwrap();
-					if let Some(txt) = drv.strip_prefix("\u{1b}[35;1m'") {
-						drv = txt;
-					}
-					if let Some(txt) = drv.strip_suffix("'\u{1b}[0m") {
-						drv = txt;
-					}
-					if let Some(txt) = drv.split("', '").next() {
-						drv = txt;
-					}
-					if let Some(pkg) = drv.strip_prefix("/nix/store/") {
-						let mut it = pkg.splitn(2, '-');
-						it.next();
-						if let Some(pkg) = it.next() {
-							drv = pkg;
-						}
-					}
-					let span = info_span!("waiting on drv", drv);
-					#[cfg(feature = "indicatif")]
-					span.pb_start();
-					self.spans.insert(id, span);
-					// Concurrent build of the same message
-				}
-				NixLog::Stop { id, .. } => {
-					self.spans.remove(&id);
-				}
-				NixLog::Result { fields, id, typ } if typ == 101 && !fields.is_empty() => {
-					if let Some(span) = self.spans.get(&id) {
-						if let LogField::String(s) = &fields[0] {
-							#[cfg(feature = "indicatif")]
-							span.pb_set_message(&process_message(s.trim()));
-							#[cfg(not(feature = "indicatif"))]
-							{
-								let _span = span.enter();
-								info!("{}", process_message(s));
-							}
-						} else {
-							warn!("bad fields: {fields:?}");
-						}
-					} else {
-						warn!("unknown result id: {id} {typ} {fields:?}");
-					}
-					// dbg!(fields, id, typ);
-				}
-				NixLog::Result { fields, id, typ } if typ == 105 && fields.len() >= 4 => {
-					if let Some(span) = self.spans.get(&id) {
-						if let [
-							LogField::Num(done),
-							LogField::Num(expected),
-							LogField::Num(_running),
-							LogField::Num(_failed),
-						] = &fields[..4]
-						{
-							#[cfg(feature = "indicatif")]
-							{
-								span.pb_set_length(*expected);
-								span.pb_set_position(*done);
-							}
-							let _ = (span, done, expected);
-						} else {
-							warn!("bad fields: {fields:?}");
-						}
-					} else {
-						// warn!("unknown result id: {id} {typ} {fields:?}");
-						// Unaccounted progress.
-					}
-					// dbg!(fields, id, typ);
-				}
-				NixLog::Result { typ, .. } if typ == 104 || typ == 106 => {
-					// Set phase, expected
-				}
-				_ => warn!("unknown log: {:?}", log),
-			};
-		} else {
-			let e = e.trim();
-			if e.starts_with("Failed tcsetattr(TCSADRAIN): ") {
-				return;
-			}
-			info!("{e}")
-		}
-	}
-}
deletedcrates/better-command/src/lib.rsdiffbeforeafterboth
--- a/crates/better-command/src/lib.rs
+++ /dev/null
@@ -1,2 +0,0 @@
-mod handler;
-pub use handler::{ClonableHandler, Handler, NixHandler, NoopHandler, PlainHandler};
modifiedcrates/fleet-base/Cargo.tomldiffbeforeafterboth
--- a/crates/fleet-base/Cargo.toml
+++ b/crates/fleet-base/Cargo.toml
@@ -5,9 +5,15 @@
 rust-version.workspace = true
 
 [dependencies]
+remowt-client.workspace = true
+remowt-endpoints.workspace = true
+remowt-fleet.workspace = true
+remowt-link-shared.workspace = true
+remowt-ui-prompt.workspace = true
+
 age.workspace = true
 anyhow.workspace = true
-better-command.workspace = true
+camino.workspace = true
 chrono.workspace = true
 clap = { workspace = true, features = ["derive"] }
 fleet-shared.workspace = true
@@ -18,15 +24,11 @@
 nix-eval.workspace = true
 nixlike.workspace = true
 nom.workspace = true
-openssh.workspace = true
 rand.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 tabled.workspace = true
 tempfile.workspace = true
-thiserror.workspace = true
 time = { workspace = true, features = ["parsing"] }
 tokio.workspace = true
-tokio-util = { workspace = true, features = ["codec"] }
-toml_edit.workspace = true
 tracing.workspace = true
deletedcrates/fleet-base/src/command.rsdiffbeforeafterboth
--- a/crates/fleet-base/src/command.rs
+++ /dev/null
@@ -1,442 +0,0 @@
-use std::{ffi::OsStr, pin, process::Stdio, sync::Arc, task::Poll};
-
-use anyhow::{Result, anyhow};
-use better_command::{Handler, NixHandler, PlainHandler};
-use futures::StreamExt;
-use itertools::Either;
-use openssh::{OverSsh, OwningCommand, Session};
-use serde::de::DeserializeOwned;
-use tokio::{io::AsyncRead, process::Command, select};
-use tokio_util::codec::{BytesCodec, FramedRead, LinesCodec};
-use tracing::debug;
-
-use crate::host::EscalationStrategy;
-
-fn escape_bash(input: &str, out: &mut String) {
-	const TO_ESCAPE: &str = "$ !\"#&'()*,;<>?[\\]^`{|}";
-	if input.chars().all(|c| !TO_ESCAPE.contains(c)) {
-		out.push_str(input);
-		return;
-	}
-	out.push('\'');
-	for (i, v) in input.split('\'').enumerate() {
-		if i != 0 {
-			out.push_str("'\"'\"'");
-		}
-		out.push_str(v);
-	}
-	out.push('\'');
-}
-fn ostoutf8(os: impl AsRef<OsStr>) -> String {
-	os.as_ref().to_str().expect("non-utf8 data").to_owned()
-}
-
-#[derive(Clone, Debug)]
-pub struct MyCommand {
-	command: String,
-	args: Vec<String>,
-	env: Vec<(String, String)>,
-	ssh_session: Option<Arc<Session>>,
-	escalation: EscalationStrategy,
-	escalate: bool,
-}
-impl MyCommand {
-	pub fn new_on(
-		escalation: EscalationStrategy,
-		cmd: impl AsRef<OsStr>,
-		session: Arc<Session>,
-	) -> Self {
-		assert!(!cmd.as_ref().is_empty());
-		Self {
-			command: ostoutf8(cmd),
-			args: vec![],
-			env: vec![],
-			ssh_session: Some(session),
-			escalation,
-			escalate: false,
-		}
-	}
-	pub fn new(escalation: EscalationStrategy, cmd: impl AsRef<OsStr>) -> Self {
-		assert!(!cmd.as_ref().is_empty());
-		Self {
-			command: ostoutf8(cmd),
-			args: vec![],
-			env: vec![],
-			ssh_session: None,
-			escalation,
-			escalate: false,
-		}
-	}
-	fn new_here(&self, cmd: impl AsRef<OsStr>) -> Self {
-		match self.ssh_session.clone() {
-			Some(ssh_session) => Self::new_on(self.escalation, cmd, ssh_session),
-			_ => Self::new(self.escalation, cmd),
-		}
-	}
-
-	fn into_args(self) -> Vec<String> {
-		let mut out = Vec::new();
-		if !self.env.is_empty() {
-			out.push("env".to_owned());
-			for (k, v) in self.env {
-				assert!(!k.contains('='));
-				out.push(format!("{k}={v}"));
-			}
-		}
-		out.push(self.command);
-		out.extend(self.args);
-		out
-	}
-
-	/// Translates environment variables into env command execution.
-	/// Required for ssh, as ssh don't allow to send environment variables (at least by default).
-	///
-	/// FIXME: Insecure, as arguments might be seen by other users on the same machine.
-	/// Figure out some way to transfer environment using stdio?
-	fn translate_env_into_env(self) -> Self {
-		if self.env.is_empty() {
-			return self;
-		}
-		let mut out = self.new_here("env");
-		for (k, v) in self.env {
-			assert!(!k.contains('='));
-			out.arg(format!("{k}={v}"));
-		}
-		out.arg(self.command);
-		out.args(self.args);
-
-		out
-	}
-	fn into_string(self) -> String {
-		let mut out = String::new();
-		if !self.env.is_empty() {
-			out.push_str("env");
-			for (k, v) in self.env {
-				out.push(' ');
-				assert!(!k.contains('='));
-				escape_bash(&k, &mut out);
-				out.push('=');
-				escape_bash(&v, &mut out);
-			}
-		}
-		if !out.is_empty() {
-			out.push(' ');
-		}
-		escape_bash(&self.command, &mut out);
-		for arg in self.args {
-			out.push(' ');
-			escape_bash(&arg, &mut out);
-		}
-		out
-	}
-	fn into_command_unchecked_local(self) -> Command {
-		let mut out = Command::new(self.command);
-		out.args(self.args);
-		for (k, v) in self.env {
-			out.env(k, v);
-		}
-		out
-	}
-	fn into_command(self) -> Result<Either<Command, openssh::OwningCommand<Arc<Session>>>> {
-		Ok(match self.ssh_session.clone() {
-			Some(session) => {
-				let cmd = self.translate_env_into_env().into_command_unchecked_local();
-				Either::Right(
-					cmd.over_ssh(session)
-						.map_err(|e| anyhow!("ssh error: {e}"))?,
-				)
-			}
-			_ => {
-				let cmd = self.into_command_unchecked_local();
-				Either::Left(cmd)
-			}
-		})
-	}
-	pub fn arg(&mut self, arg: impl AsRef<OsStr>) -> &mut Self {
-		let arg = arg.as_ref();
-		self.args.push(ostoutf8(arg));
-		self
-	}
-	pub fn eqarg(&mut self, arg: impl AsRef<OsStr>, value: impl AsRef<OsStr>) -> &mut Self {
-		let arg = arg.as_ref();
-		let value = value.as_ref();
-		let arg = ostoutf8(arg);
-		let value = ostoutf8(value);
-		self.arg(format!("{arg}={value}"));
-		self
-	}
-	pub fn comparg(&mut self, arg: impl AsRef<OsStr>, value: impl AsRef<OsStr>) -> &mut Self {
-		self.arg(arg);
-		self.arg(value);
-		self
-	}
-	pub fn env(&mut self, name: impl AsRef<str>, value: impl AsRef<str>) -> &mut Self {
-		self.env
-			.push((name.as_ref().to_owned(), value.as_ref().to_owned()));
-		self
-	}
-	pub fn args<V: AsRef<OsStr>>(&mut self, args: impl IntoIterator<Item = V>) -> &mut Self {
-		for arg in args.into_iter() {
-			let arg = arg.as_ref();
-			self.args.push(ostoutf8(arg));
-		}
-		self
-	}
-	pub fn sudo(mut self) -> Self {
-		self.escalate = true;
-		self
-	}
-	fn wrap_sudo_if_needed(self) -> Self {
-		if !self.escalate {
-			return self;
-		}
-		match self.escalation {
-			EscalationStrategy::Su => {
-				let mut out = self.new_here("su");
-				out.arg("-c").arg(self.into_string());
-				out
-			}
-			EscalationStrategy::Sudo => {
-				let mut out = self.new_here("sudo");
-				out.args(self.into_args());
-				out
-			}
-			EscalationStrategy::Run0 => {
-				// run0 wants interactive authentication by default.
-				let mut run0 = self.new_here("run0");
-				let mut out = self.new_here("script");
-
-				// Red backgrounds messes with fleet formatting
-				run0.arg("--background=");
-				run0.args(self.into_args());
-
-				out.arg("-q");
-				out.arg("/dev/null");
-				out.arg("-c");
-				out.arg(run0.into_string());
-				dbg!(&out);
-				out
-			}
-		}
-	}
-
-	pub async fn run(self) -> Result<()> {
-		let str = self.clone().into_string();
-		let cmd = self.wrap_sudo_if_needed().into_command()?;
-		match cmd {
-			Either::Left(cmd) => run_nix_inner(str, cmd, &mut PlainHandler).await?,
-			Either::Right(cmd) => run_nix_inner_ssh(str, cmd, &mut PlainHandler).await?,
-		};
-		Ok(())
-	}
-	pub async fn run_string(self) -> Result<String> {
-		let bytes = self.run_bytes().await?;
-		Ok(String::from_utf8(bytes)?)
-	}
-	pub async fn run_value<T: DeserializeOwned>(self) -> Result<T> {
-		let v = self.run_string().await?;
-		Ok(serde_json::from_str(&v)?)
-	}
-	pub async fn run_bytes(self) -> Result<Vec<u8>> {
-		let str = self.clone().into_string();
-		let cmd = self.wrap_sudo_if_needed().into_command()?;
-		let v = match cmd {
-			Either::Left(cmd) => run_nix_inner_stdout(str, cmd, &mut PlainHandler).await?,
-			Either::Right(cmd) => run_nix_inner_stdout_ssh(str, cmd, &mut PlainHandler).await?,
-		};
-		Ok(v)
-	}
-
-	pub async fn run_nix_string(mut self) -> Result<String> {
-		let str = self.clone().into_string();
-		self.arg("--log-format").arg("internal-json");
-		let cmd = self.wrap_sudo_if_needed().into_command()?;
-		let bytes = match cmd {
-			Either::Left(cmd) => run_nix_inner_stdout(str, cmd, &mut NixHandler::default()).await?,
-			Either::Right(cmd) => {
-				run_nix_inner_stdout_ssh(str, cmd, &mut NixHandler::default()).await?
-			}
-		};
-		Ok(String::from_utf8(bytes)?)
-	}
-	pub async fn run_nix(mut self) -> Result<()> {
-		let str = self.clone().into_string();
-		self.arg("--log-format").arg("internal-json");
-		let cmd = self.wrap_sudo_if_needed().into_command()?;
-		match cmd {
-			Either::Left(mut cmd) => {
-				cmd.stdout(Stdio::inherit());
-				run_nix_inner(str, cmd, &mut NixHandler::default()).await
-			}
-			Either::Right(mut cmd) => {
-				cmd.stdout(openssh::Stdio::inherit());
-				run_nix_inner_ssh(str, cmd, &mut NixHandler::default()).await
-			}
-		}
-	}
-}
-
-struct EmptyAsyncRead;
-impl AsyncRead for EmptyAsyncRead {
-	fn poll_read(
-		self: std::pin::Pin<&mut Self>,
-		_cx: &mut std::task::Context<'_>,
-		_buf: &mut tokio::io::ReadBuf<'_>,
-	) -> Poll<std::io::Result<()>> {
-		Poll::Pending
-	}
-}
-
-async fn run_nix_inner_stdout(
-	str: String,
-	cmd: Command,
-	handler: &mut dyn Handler,
-) -> Result<Vec<u8>> {
-	Ok(run_nix_inner_raw(str, cmd, true, handler, None)
-		.await?
-		.expect("has out"))
-}
-async fn run_nix_inner(str: String, cmd: Command, handler: &mut dyn Handler) -> Result<()> {
-	let v = run_nix_inner_raw(str, cmd, false, handler, None).await?;
-	assert!(v.is_none());
-	Ok(())
-}
-async fn run_nix_inner_stdout_ssh(
-	str: String,
-	cmd: OwningCommand<Arc<Session>>,
-	handler: &mut dyn Handler,
-) -> Result<Vec<u8>> {
-	Ok(run_nix_inner_raw_ssh(str, cmd, true, handler, None)
-		.await?
-		.expect("has out"))
-}
-async fn run_nix_inner_ssh(
-	str: String,
-	cmd: OwningCommand<Arc<Session>>,
-	handler: &mut dyn Handler,
-) -> Result<()> {
-	let v = run_nix_inner_raw_ssh(str, cmd, false, handler, None).await?;
-	assert!(v.is_none());
-	Ok(())
-}
-
-async fn run_nix_inner_raw(
-	str: String,
-	mut cmd: Command,
-	want_stdout: bool,
-	err_handler: &mut dyn Handler,
-	mut out_handler: Option<&mut dyn Handler>,
-) -> Result<Option<Vec<u8>>> {
-	cmd.stderr(Stdio::piped());
-	cmd.stdout(Stdio::piped());
-	debug!("running command {str:?} on local");
-	let mut child = cmd.spawn()?;
-	let mut stderr = child.stderr.take().unwrap();
-	let stdout = child.stdout.take().unwrap();
-	let mut err = FramedRead::new(&mut stderr, LinesCodec::new());
-	let mut out: Option<Box<dyn AsyncRead + Unpin + Send>> = Some(Box::new(stdout));
-	let mut ob = want_stdout
-		.then(|| out.take().unwrap())
-		.unwrap_or_else(|| Box::new(EmptyAsyncRead));
-	let mut ol = (!want_stdout)
-		.then(|| out.take().unwrap())
-		.unwrap_or_else(|| Box::new(EmptyAsyncRead));
-	let mut ob = FramedRead::new(&mut ob, BytesCodec::new());
-	let mut ol = FramedRead::new(&mut ol, LinesCodec::new());
-
-	// while let Some(line) = read.next().await? {}
-
-	let mut out_buf = if want_stdout { Some(vec![]) } else { None };
-	loop {
-		select! {
-			biased;
-
-			Some(e) = err.next() => {
-				let e = e?;
-				err_handler.handle_line(&e);
-			},
-			Some(o) = ob.next() => {
-				out_buf.as_mut().expect("stdout == wants_stdout").extend_from_slice(&o?);
-			},
-			Some(o) = ol.next() => {
-				let o = o?;
-				if let Some(out) = out_handler.as_mut() {
-					out.handle_line(&o)
-				} else {
-					err_handler.handle_line(&o)
-				}
-				// out_handler.handle_info(&o);
-			},
-			code = child.wait() => {
-				let code = code?;
-				if !code.success() {
-					anyhow::bail!("command '{str}' failed with status {}", code);
-				}
-				break;
-			}
-		}
-	}
-
-	Ok(out_buf)
-}
-async fn run_nix_inner_raw_ssh(
-	str: String,
-	mut cmd: OwningCommand<Arc<Session>>,
-	want_stdout: bool,
-	err_handler: &mut dyn Handler,
-	mut out_handler: Option<&mut dyn Handler>,
-) -> Result<Option<Vec<u8>>> {
-	debug!("running command {str:?} over ssh");
-	cmd.stderr(openssh::Stdio::piped());
-	cmd.stdout(openssh::Stdio::piped());
-	let mut child = cmd.spawn().await?;
-	let mut stderr = child.stderr().take().unwrap();
-	let stdout = child.stdout().take().unwrap();
-	let mut err = FramedRead::new(&mut stderr, LinesCodec::new());
-	let mut out: Option<Box<dyn AsyncRead + Unpin + Send>> = Some(Box::new(stdout));
-	let mut ob = want_stdout
-		.then(|| out.take().unwrap())
-		.unwrap_or_else(|| Box::new(EmptyAsyncRead));
-	let mut ol = (!want_stdout)
-		.then(|| out.take().unwrap())
-		.unwrap_or_else(|| Box::new(EmptyAsyncRead));
-	let mut ob = FramedRead::new(&mut ob, BytesCodec::new());
-	let mut ol = FramedRead::new(&mut ol, LinesCodec::new());
-
-	// while let Some(line) = read.next().await? {}
-
-	let mut out_buf = if want_stdout { Some(vec![]) } else { None };
-
-	let mut wait_future = pin::pin!(child.wait());
-	loop {
-		select! {
-			biased;
-
-			Some(e) = err.next() => {
-				let e = e?;
-				err_handler.handle_line(&e);
-			},
-			Some(o) = ob.next() => {
-				out_buf.as_mut().expect("stdout == wants_stdout").extend_from_slice(&o?);
-			},
-			Some(o) = ol.next() => {
-				let o = o?;
-				if let Some(out) = out_handler.as_mut() {
-					out.handle_line(&o)
-				} else {
-					err_handler.handle_line(&o)
-				}
-				// out_handler.handle_info(&o);
-			},
-			code = &mut wait_future => {
-				let code = code?;
-				if !code.success() {
-					anyhow::bail!("command '{str}' failed with status {}", code);
-				}
-				break;
-			}
-		}
-	}
-
-	Ok(out_buf)
-}
modifiedcrates/fleet-base/src/deploy.rsdiffbeforeafterboth
--- a/crates/fleet-base/src/deploy.rs
+++ b/crates/fleet-base/src/deploy.rs
@@ -1,8 +1,13 @@
-use std::{path::PathBuf, time::Duration};
+use std::time::Duration;
 
 use anyhow::{Context as _, Result, anyhow, bail};
+use camino::Utf8PathBuf;
 use clap::ValueEnum;
 use itertools::Itertools;
+use remowt_endpoints::fs::FsClient;
+use remowt_endpoints::systemd::SystemdClient;
+use remowt_fleet::NixClient;
+use remowt_link_shared::BifConfig;
 use tokio::time::sleep;
 use tracing::{Instrument as _, error, info, info_span, warn};
 
@@ -59,10 +64,11 @@
 pub async fn deploy_task(
 	action: DeployAction,
 	host: &ConfigHost,
-	built: PathBuf,
+	built: Utf8PathBuf,
 	specialisation: Option<String>,
 	disable_rollback: bool,
 ) -> Result<()> {
+	let remowt = host.remowt().await?;
 	let deploy_kind = host.deploy_kind().await?;
 	if (deploy_kind == DeployKind::NixosInstall || deploy_kind == DeployKind::NixosLustrate)
 		&& !matches!(action, DeployAction::Boot | DeployAction::Upload)
@@ -86,7 +92,7 @@
 			generation.id, generation.datetime
 		);
 		{
-			let mut cmd = host.cmd("sh").await?;
+			let mut cmd = remowt.cmd("sh");
 			cmd.arg("-c").arg(format!("mark=$(mktemp -p /etc -t fleet_rollback_marker.XXXXX) && echo -n {} > $mark && mv --no-clobber $mark /etc/fleet_rollback_marker", generation.id));
 			if let Err(e) = cmd.sudo().run().await {
 				error!("failed to set rollback marker: {e}");
@@ -104,7 +110,7 @@
 		// if we fail to perform generation switch in time, then we will still call the activation script, and this may break something.
 		// Anyway, reboot will still help in this case.
 		if action.should_schedule_rollback_run() {
-			let mut cmd = host.cmd("systemd-run").await?;
+			let mut cmd = remowt.cmd("systemd-run");
 			cmd.comparg("--on-active", "3min")
 				.comparg("--unit", "rollback-watchdog-run")
 				.arg("systemctl")
@@ -116,14 +122,20 @@
 			}
 		}
 	}
+
+	let remowt = host.remowt().await?;
+	let fs = remowt.endpoints::<FsClient<_>>();
 	if deploy_kind == DeployKind::NixosLustrate {
 		// Fleet could also create this file, but as this operation is potentially disruptive,
 		// make user do it themself.
-		if !host.file_exists("/etc/NIXOS_LUSTRATE").await? {
+		if !fs
+			.file_exists(Utf8PathBuf::from("/etc/NIXOS_LUSTRATE"))
+			.await?
+		{
 			bail!("/etc/NIXOS_LUSTRATE should be created on remote host");
 		}
 		// Wanted by NixOS to recognize the system as NixOS.
-		let mut cmd = host.cmd("touch").await?;
+		let mut cmd = remowt.cmd("touch");
 		cmd.arg("/etc/NIXOS");
 		cmd.sudo().run().await.context("creating /etc/NIXOS")?;
 	}
@@ -131,7 +143,7 @@
 		info!(
 			"running nixos-install to switch profile, install bootloader, and perform activation"
 		);
-		let mut cmd = host.cmd("nixos-install").await?;
+		let mut cmd = remowt.cmd("nixos-install");
 		cmd.arg("--system").arg(&built).args([
 			// Channels here aren't fleet host system channels, but channels embedded in installation cd, which might be old.
 			// It is possible to copy host channels, but I would prefer non-flake nix just to be unsupported.
@@ -147,32 +159,38 @@
 		if action.should_switch_profile() && !failed {
 			info!("switching system profile generation");
 
-			// To avoid even more problems, using nixos-install for now.
-			// // nix build is unable to work with --store argument for some reason, and nix until 2.26 didn't support copy with --profile argument,
-			// // falling back to using nix-env command
-			// // After stable NixOS starts using 2.26 - use `nix --store /mnt copy --from /mnt --profile ...` here, and instead of nix build below.
-			// let mut cmd = host.cmd("nix-env").await?;
-			// cmd.args([
-			// 	"--store",
-			// 	"/mnt",
-			// 	"--profile",
-			// 	"/mnt/nix/var/nix/profiles/system",
-			// 	"--set",
-			// ])
-			// .arg(&built);
-			// if let Err(e) = cmd.sudo().run_nix().await {
-			// 	error!("failed to switch system profile generation: {e}");
-			// 	failed = true;
-			// }
-			// It would also be possible to update profile atomically during copy:
-			// https://github.com/NixOS/nix/pull/11657
-			let mut cmd = host.nix_cmd().await?;
-			cmd.arg("build");
-			cmd.comparg("--profile", "/nix/var/nix/profiles/system");
-			cmd.arg(&built);
-			if let Err(e) = cmd.sudo().run_nix().await {
-				error!("failed to switch system profile generation: {e}");
-				failed = true;
+			match host.ensure_nix_plugin().await {
+				Ok(plugin_id) => {
+					let nix_elevated = remowt.plugin_endpoints::<NixClient<_>>(plugin_id);
+					// To avoid even more problems, using nixos-install for now.
+					// // nix build is unable to work with --store argument for some reason, and nix until 2.26 didn't support copy with --profile argument,
+					// // falling back to using nix-env command
+					// // After stable NixOS starts using 2.26 - use `nix --store /mnt copy --from /mnt --profile ...` here, and instead of nix build below.
+					// let mut cmd = host.cmd("nix-env").await?;
+					// cmd.args([
+					// 	"--store",
+					// 	"/mnt",
+					// 	"--profile",
+					// 	"/mnt/nix/var/nix/profiles/system",
+					// 	"--set",
+					// ])
+					// .arg(&built);
+					// if let Err(e) = cmd.sudo().run_nix().await {
+					// 	error!("failed to switch system profile generation: {e}");
+					// 	failed = true;
+					// }
+					if let Err(e) = nix_elevated
+						.switch_profile("/nix/var/nix/profiles/system".to_owned(), built.clone())
+						.await
+					{
+						error!("failed to switch system profile generation: {e}");
+						failed = true;
+					}
+				}
+				Err(e) => {
+					failed = true;
+					error!("failed to enable nix plugin: {e:?}");
+				}
 			}
 		}
 
@@ -189,7 +207,7 @@
 				built.clone()
 			};
 			let switch_script = specialised.join("bin/switch-to-configuration");
-			let mut cmd = host.cmd("systemd-run").in_current_span().await?;
+			let mut cmd = remowt.cmd("systemd-run");
 			if deploy_kind == DeployKind::NixosLustrate {
 				cmd.arg("--setenv=NIXOS_INSTALL_BOOTLOADER=1");
 			}
@@ -209,12 +227,14 @@
 		}
 	}
 	if action.should_create_rollback_marker() {
+		let elevated_systemd = remowt.run0_endpoints::<SystemdClient<BifConfig>>().await?;
+		let elevated_fs = remowt.run0_endpoints::<FsClient<BifConfig>>().await?;
 		if !disable_rollback {
 			if failed {
 				if action.should_schedule_rollback_run() {
 					info!("executing rollback");
-					if let Err(e) = host
-						.systemctl_start("rollback-watchdog.service")
+					if let Err(e) = elevated_systemd
+						.start("rollback-watchdog.service".to_owned())
 						.instrument(info_span!("rollback"))
 						.await
 					{
@@ -223,8 +243,8 @@
 				}
 			} else {
 				info!("trying to mark upgrade as successful");
-				if let Err(e) = host
-					.rm_file("/etc/fleet_rollback_marker", true)
+				if let Err(e) = elevated_fs
+					.rm_file(Utf8PathBuf::from("/etc/fleet_rollback_marker"))
 					.in_current_span()
 					.await
 				{
@@ -234,16 +254,22 @@
 				}
 			}
 			info!("disarming watchdog, just in case");
-			if let Err(_e) = host.systemctl_stop("rollback-watchdog.timer").await {
+			if let Err(_e) = elevated_systemd
+				.stop("rollback-watchdog.timer".to_owned())
+				.await
+			{
 				// It is ok, if there was no reboot - then timer might not be running.
 			}
 			if action.should_schedule_rollback_run() {
-				if let Err(e) = host.systemctl_stop("rollback-watchdog-run.timer").await {
+				if let Err(e) = elevated_systemd
+					.stop("rollback-watchdog-run.timer".to_owned())
+					.await
+				{
 					error!("failed to disarm rollback run: {e}");
 				}
 			}
-		} else if let Err(_e) = host
-			.rm_file("/etc/fleet_rollback_marker", true)
+		} else if let Err(_e) = elevated_fs
+			.rm_file(Utf8PathBuf::from("/etc/fleet_rollback_marker"))
 			.in_current_span()
 			.await
 		{
@@ -257,8 +283,8 @@
 	config: &Config,
 	host: &ConfigHost,
 	location: GenerationStorage,
-	generation: PathBuf,
-) -> Result<PathBuf> {
+	generation: Utf8PathBuf,
+) -> Result<Utf8PathBuf> {
 	if matches!(location, GenerationStorage::Pusher) {
 		bail!("pusher is not enabled in this version of fleet");
 	}
@@ -273,11 +299,11 @@
 				}
 				Err(e) if tries < 3 => {
 					tries += 1;
-					warn!("copy failure ({}/3): {}", tries, e);
+					warn!("copy failure ({}/3): {:#}", tries, e);
 					sleep(Duration::from_millis(5000)).await;
 				}
 				Err(e) => {
-					bail!("upload failed: {e}");
+					bail!("upload failed: {e:#}");
 				}
 			}
 		}
modifiedcrates/fleet-base/src/fleetdata.rsdiffbeforeafterboth
--- a/crates/fleet-base/src/fleetdata.rs
+++ b/crates/fleet-base/src/fleetdata.rs
@@ -144,10 +144,6 @@
 	pub generation_data: Value,
 }
 
-fn is_false(b: &bool) -> bool {
-	*b == false
-}
-
 #[derive(Serialize, Deserialize, Clone, Debug, PartialOrd, Ord, PartialEq, Eq)]
 #[repr(transparent)]
 pub struct SecretOwner(String);
@@ -291,7 +287,7 @@
 	}
 
 	// Which one is easier to access
-	return a.owners.len().cmp(&b.owners.len());
+	a.owners.len().cmp(&b.owners.len())
 }
 
 impl FleetSecretDistributions {
modifiedcrates/fleet-base/src/host.rsdiffbeforeafterboth
--- a/crates/fleet-base/src/host.rs
+++ b/crates/fleet-base/src/host.rs
@@ -1,30 +1,33 @@
 use std::{
 	collections::{BTreeMap, BTreeSet, HashSet},
-	ffi::{OsStr, OsString},
-	fmt::Display,
+	future::Future,
 	io::Write,
 	ops::Deref,
 	path::PathBuf,
+	pin::Pin,
 	str::FromStr,
-	sync::{Arc, Mutex, MutexGuard, OnceLock},
+	sync::{Arc, OnceLock},
 };
 
 use anyhow::{Context, Result, anyhow, bail, ensure};
+use camino::{Utf8Path, Utf8PathBuf};
 use chrono::{DateTime, Utc};
 use fleet_shared::SecretData;
-use nix_eval::{Value, nix_go, nix_go_json, util::assert_warn};
-use openssh::{ControlPersist, SessionBuilder};
-use serde::de::DeserializeOwned;
+use nix_eval::{Store, Value, nix_go, nix_go_json, util::assert_warn};
+use remowt_client::{AgentBundle, Remowt};
+use remowt_endpoints::fs::FsClient;
+use remowt_link_shared::Address;
+use remowt_ui_prompt::auto::AutoPrompter;
+use remowt_ui_prompt::bifrost::PromptEndpoints;
+use remowt_ui_prompt::{PrependSourcePrompter, Source};
 use tabled::Tabled;
 use tempfile::NamedTempFile;
-use time::{UtcDateTime, format_description};
-use tracing::warn;
+use time::UtcDateTime;
+use tokio::task::spawn_blocking;
+use tracing::{info, warn};
 
-use crate::{
-	command::MyCommand,
-	fleetdata::{
-		FleetData, FleetSecretData, FleetSecretDistribution, FleetSecretPart, SecretOwner,
-	},
+use crate::fleetdata::{
+	FleetData, FleetSecretData, FleetSecretDistribution, FleetSecretPart, SecretOwner,
 };
 
 pub struct FleetConfigInternals {
@@ -36,7 +39,6 @@
 	/// builtins.currentSystem
 	pub local_system: String,
 	pub data: Arc<FleetData>,
-	pub nix_args: Vec<OsString>,
 	/// fleet_config.config
 	pub config_field: Value,
 	/// flake.output
@@ -48,6 +50,8 @@
 	pub default_pkgs: Value,
 	/// inputs.nixpkgs
 	pub nixpkgs: Value,
+
+	pub local_host: OnceLock<Arc<ConfigHost>>,
 }
 
 // TODO: Make field not pub
@@ -62,13 +66,6 @@
 	}
 }
 
-#[derive(Clone, Copy, Debug)]
-pub enum EscalationStrategy {
-	Sudo,
-	Run0,
-	Su,
-}
-
 #[derive(Clone, PartialEq, Copy, Debug)]
 pub enum DeployKind {
 	/// NixOS => NixOS managed by fleet
@@ -115,9 +112,26 @@
 
 	// TODO: Move command helpers away with connectivity refactor
 	pub local: bool,
-	pub session: OnceLock<Arc<openssh::Session>>,
+	pub remowt: OnceLock<Remowt>,
+	nix_store: OnceLock<Arc<Store>>,
+	nix_plugin: tokio::sync::OnceCell<()>,
 }
 
+const NIX_PLUGIN_ID: u16 = 2;
+
+fn agents_dir() -> Result<PathBuf> {
+	std::env::var_os("REMOWT_AGENTS_DIR")
+		.map(PathBuf::from)
+		.or_else(|| option_env!("REMOWT_AGENTS_DIR").map(PathBuf::from))
+		.ok_or_else(|| {
+			anyhow!("no remowt-agents bundle; set REMOWT_AGENTS_DIR to a remowt-agents output")
+		})
+}
+
+fn agent_bundle() -> Result<AgentBundle> {
+	AgentBundle::from_dir(agents_dir()?)
+}
+
 #[derive(Debug, Clone, Copy)]
 pub enum GenerationStorage {
 	Deployer,
@@ -143,7 +157,7 @@
 	#[tabled(rename = "Created at")]
 	pub datetime: UtcDateTime,
 	#[tabled(format = "{:?}")]
-	pub store_path: PathBuf,
+	pub store_path: Utf8PathBuf,
 	#[tabled(skip)]
 	pub location: GenerationStorage,
 }
@@ -153,68 +167,36 @@
 	}
 }
 
-fn parse_generation_line(g: &str) -> Option<Generation> {
-	let mut parts = g.split_whitespace();
-	let id = parts.next()?;
-	let id: u32 = id.parse().ok()?;
-	let date = parts.next()?;
-	let time = parts.next()?;
-	let current = if let Some(current) = parts.next() {
-		if current == "(current)" {
-			Some(true)
-		} else {
-			None
-		}
-	} else {
-		Some(false)
-	};
-	let current = current?;
-	if parts.next().is_some() {
-		warn!("unexpected text after generation: {g}");
-	}
-
-	let format = format_description::parse("[year]-[month]-[day] [hour]:[minute]:[second]")
-		.expect("valid format");
-	let datetime = UtcDateTime::parse(&format!("{date} {time}"), &format).ok()?;
-
-	Some(Generation {
-		id,
-		current,
-		datetime,
-		store_path: PathBuf::new(),
-		location: GenerationStorage::Machine,
-	})
-}
-// TODO: Move command helpers away with connectivity refactor
 impl ConfigHost {
 	pub async fn list_generations(&self, profile: &str) -> Result<Vec<Generation>> {
-		let mut cmd = self.cmd("nix-env").await?;
-		cmd.comparg("--profile", format!("/nix/var/nix/profiles/{profile}"))
-			.arg("--list-generations")
-			.env("TZ", "UTC");
-		// Sudo is required because --list-generations tries to acquire profile lock
-		let data = cmd.sudo().run_string().await?;
-		let mut generations = data
-			.split('\n')
-			.map(|e| e.trim())
-			.filter(|&l| !l.is_empty())
-			.filter_map(|g| {
-				let generation = parse_generation_line(g);
-				if generation.is_none() {
-					warn!("bad generation: {g}");
-				};
-				generation
+		let plugin_id = self.ensure_nix_plugin().await?;
+		let nix = self
+			.remowt()
+			.await?
+			.plugin_endpoints::<remowt_fleet::NixClient<_>>(plugin_id);
+		let raw = nix
+			.list_generations(profile.to_owned())
+			.await
+			.map_err(|e| anyhow!("{e:?}"))?
+			.map_err(|e| anyhow!("{e}"))?;
+		raw.into_iter()
+			.map(|g| {
+				let id: u32 =
+					g.id.try_into()
+						.with_context(|| format!("generation id {} doesn't fit in u32", g.id))?;
+				let datetime = UtcDateTime::from_unix_timestamp(g.creation_time_unix)
+					.with_context(|| {
+						format!("invalid generation timestamp {}", g.creation_time_unix)
+					})?;
+				Ok(Generation {
+					id,
+					current: g.current,
+					datetime,
+					store_path: g.store_path,
+					location: GenerationStorage::Machine,
+				})
 			})
-			.collect::<Vec<_>>();
-		for ele in generations.iter_mut() {
-			let mut cmd = self.cmd("readlink").await?;
-			cmd.arg("--")
-				.arg(format!("/nix/var/nix/profiles/{profile}-{}-link", ele.id));
-			let path = cmd.run_string().await?;
-			ele.store_path = PathBuf::from(path.trim_end_matches("\n"));
-		}
-
-		Ok(generations)
+			.collect()
 	}
 
 	pub fn set_session_destination(&self, dest: String) {
@@ -236,7 +218,9 @@
 		if let Some(kind) = self.deploy_kind.get() {
 			return Ok(*kind);
 		}
-		let is_fleet_managed = match self.file_exists("/etc/FLEET_HOST").await {
+		let remowt = self.remowt().await?;
+		let fs = remowt.endpoints::<FsClient<_>>();
+		let is_fleet_managed = match fs.file_exists(Utf8PathBuf::from("/etc/FLEET_HOST")).await {
 			Ok(v) => v,
 			Err(e) => {
 				bail!("failed to query remote system kind: {e}");
@@ -252,6 +236,7 @@
 					1. manually create /etc/FLEET_HOST file on the target host,
 					2. use ?deploy_kind=fleet host argument if you're upgrading from older version of fleet
 					3. use ?deploy_kind=upgrade_to_fleet if you're upgrading from plain nixos to fleet-managed nixos
+				for installation use ?deploy_kind=nixos_install / ?deploy_kind=nixos_lustrate 
 			"}
 			);
 		}
@@ -259,139 +244,101 @@
 		let _ = self.deploy_kind.set(DeployKind::Fleet);
 		Ok(*self.deploy_kind.get().expect("deploy kind is just set"))
 	}
-	pub async fn escalation_strategy(&self) -> Result<EscalationStrategy> {
-		// Prefer sudo, as run0 has some gotchas with polkit
-		// and too many repeating prompts.
-		if (self.find_in_path("sudo").await).is_ok() {
-			return Ok(EscalationStrategy::Sudo);
-		}
-		if (self.find_in_path("run0").await).is_ok() {
-			return Ok(EscalationStrategy::Run0);
+	async fn connection(&self) -> Result<Remowt> {
+		if let Some(conn) = self.remowt.get() {
+			return Ok(conn.clone());
 		}
-		Ok(EscalationStrategy::Su)
-	}
-	async fn open_session(&self) -> Result<Arc<openssh::Session>> {
-		assert!(!self.local, "do not open ssh connection to local session");
-		// FIXME: TOCTOU
-		if let Some(session) = &self.session.get() {
-			return Ok((*session).clone());
+		let bundle = agent_bundle()?;
+		let conn = if self.local {
+			Remowt::connect_local(&bundle)
+				.await
+				.context("starting local remowt agent")?
+		} else {
+			let dest = self
+				.session_destination
+				.get()
+				.cloned()
+				.unwrap_or_else(|| self.name.clone());
+			Remowt::connect(&dest, &bundle)
+				.await
+				.map_err(|e| anyhow!("remowt error while connecting to {}: {e:#?}", self.name))?
 		};
-		let mut session = SessionBuilder::default();
-		session.control_persist(ControlPersist::ClosedAfterInitialConnection);
-
-		let dest = self.session_destination.get().unwrap_or(&self.name);
-		let session = session
-			.connect(&dest)
-			.await
-			.map_err(|e| anyhow!("ssh error while connecting to {}: {e:#?}", self.name))?;
-		let session = Arc::new(session);
-		self.session.set(session.clone()).expect("TOCTOU happened");
-		Ok(session)
-	}
-	pub async fn mktemp_dir(&self) -> Result<String> {
-		let mut cmd = self.cmd("mktemp").await?;
-		cmd.arg("-d");
-		let path = cmd.run_string().await?;
-		Ok(path.trim_end().to_owned())
-	}
-	pub async fn file_exists(&self, path: impl AsRef<OsStr>) -> Result<bool> {
-		let mut cmd = self.cmd("sh").await?;
-		cmd.arg("-c")
-			.arg("test -e \"$1\" && echo true || echo false")
-			.arg("_")
-			.arg(path);
-		cmd.run_value().await
-	}
-	pub async fn read_file_bin(&self, path: impl AsRef<OsStr>) -> Result<Vec<u8>> {
-		let mut cmd = self.cmd("cat").await?;
-		cmd.arg(path);
-		cmd.run_bytes().await
-	}
-	pub async fn read_file_text(&self, path: impl AsRef<OsStr>) -> Result<String> {
-		let mut cmd = self.cmd("cat").await?;
-		cmd.arg(path);
-		cmd.run_string().await
-	}
-	pub async fn read_dir(&self, path: impl AsRef<OsStr>) -> Result<Vec<String>> {
-		let mut cmd = self.cmd("ls").await?;
-		cmd.arg(path);
-		let out = cmd.run_string().await?;
-		let mut lines = out.split('\n');
-		if let Some(last) = lines.next_back() {
-			ensure!(last.is_empty(), "output of ls should end with newline");
-		}
-		Ok(lines.map(ToOwned::to_owned).collect())
-	}
-	#[allow(dead_code)]
-	pub async fn read_file_json<D: DeserializeOwned>(&self, path: impl AsRef<OsStr>) -> Result<D> {
-		let text = self.read_file_text(path).await?;
-		Ok(serde_json::from_str(&text)?)
-	}
-	pub async fn read_env(&self, env: &str) -> Result<String> {
-		let mut cmd = self.cmd("printenv").await?;
-		cmd.arg(env);
-		cmd.run_string().await
-	}
-	pub async fn find_in_path(&self, command: &str) -> Result<String> {
-		// // `which` is not a part of coreutils, and it might not exist on machine.
-		// let path = self.read_env("PATH").await?;
-		// // Assuming delimiter is :, we don't work with windows host, this check will be much
-		// // more sophisticated in remowt backend (and quicker, since actual PATH search will be done on remote machine)
-		// for ele in path.split(':') {
-		// 	let test_path = format!("{ele}/{cmd}");
-		// 	test -x etc
-		// }
-		// let mut cmd = self.cmd("printenv").await?;
-		// cmd.arg(env);
-		// Ok(cmd.run_string().await?)
-		// Assuming this is an environment issue if which doesn't exist, will be fixed with remowt.
-		let mut cmd = self
-			.cmd_escalation(
-				// Not used
-				EscalationStrategy::Su,
-				"which",
-			)
-			.await?;
-		cmd.arg(command);
-		cmd.run_string().await
+		PromptEndpoints(PrependSourcePrompter {
+			prompter: AutoPrompter::new().await,
+			source: if self.local {
+				vec![]
+			} else {
+				vec![Source(std::borrow::Cow::Owned(format!(
+					"ssh host: {}",
+					self.name
+				)))]
+			},
+			description: "".to_owned(),
+		})
+		.register_endpoints(&mut conn.rpc());
+		let _ = self.remowt.set(conn);
+		Ok(self.remowt.get().expect("just set").clone())
 	}
-	pub async fn read_file_value<D: FromStr>(&self, path: impl AsRef<OsStr>) -> Result<D>
-	where
-		<D as FromStr>::Err: Display,
-	{
-		let text = self.read_file_text(path).await?;
-		D::from_str(&text).map_err(|e| anyhow!("failed to parse value: {e}"))
+
+	/// Client for this host's unprivileged agent.
+	pub async fn remowt(&self) -> Result<Remowt> {
+		Ok(self.connection().await?)
 	}
-	pub async fn cmd(&self, cmd: impl AsRef<OsStr>) -> Result<MyCommand> {
-		self.cmd_escalation(self.escalation_strategy().await?, cmd)
-			.await
+
+	pub fn ensure_nix_plugin(&self) -> Pin<Box<dyn Future<Output = Result<u16>> + Send + '_>> {
+		Box::pin(async {
+			self.nix_plugin
+				.get_or_try_init(|| async {
+					let pkgs = self.pkgs()?;
+					let name = "remowt-plugin-fleet";
+					let plugin = nix_go!(pkgs[{ name }]);
+					let built = plugin
+						.build("out")
+						.context("failed to build the fleet nix plugin")?;
+					let copied = self
+						.remote_derivation(&built)
+						.await
+						.context("failed to copy the fleet nix plugin to the host store")?;
+					let bin = copied.join("bin/remowt-plugin-fleet");
+					self.remowt()
+						.await?
+						.run0_load_plugin_path(NIX_PLUGIN_ID, bin.as_str())
+						.await
+						.context("failed to load the fleet nix plugin")?;
+					self.remowt()
+						.await?
+						.rpc()
+						.wait_for_connection_to(Address::Plugin(NIX_PLUGIN_ID))
+						.await
+						.map_err(|e| anyhow!("failed to wait for plugin"))?;
+					anyhow::Ok(())
+				})
+				.await?;
+			Ok(NIX_PLUGIN_ID)
+		})
 	}
-	pub async fn cmd_escalation(
-		&self,
-		escalation: EscalationStrategy,
-		cmd: impl AsRef<OsStr>,
-	) -> Result<MyCommand> {
-		if self.local {
-			Ok(MyCommand::new(escalation, cmd))
-		} else {
-			let session = self.open_session().await?;
-			Ok(MyCommand::new_on(escalation, cmd, session))
+
+	async fn nix_store(&self) -> Result<Arc<Store>> {
+		if let Some(store) = self.nix_store.get() {
+			return Ok(store.clone());
 		}
-	}
-	pub async fn nix_cmd(&self) -> Result<MyCommand> {
-		let mut nix = self.cmd("nix").await?;
-		nix.args([
-			"--extra-experimental-features",
-			"nix-command",
-			"--extra-experimental-features",
-			"flakes",
-		]);
-		Ok(nix)
+		let conn = self.connection().await?;
+		let socket = match self.deploy_kind().await? {
+			DeployKind::NixosInstall => {
+				remowt_fleet::nix_store_socket(conn, "/mnt?require-sigs=false").await?
+			}
+			_ => remowt_fleet::nix_store_socket(conn, "auto").await?,
+		};
+		let uri = format!("unix://{}", socket.display());
+		let store = Arc::new(Store::open(&uri)?);
+		let _ = self.nix_store.set(store);
+		Ok(self.nix_store.get().expect("just set").clone())
 	}
 
 	pub async fn decrypt(&self, data: SecretData) -> Result<Vec<u8>> {
 		ensure!(data.encrypted, "secret is not encrypted");
-		let mut cmd = self.cmd("fleet-install-secrets").await?;
+		let remowt = self.remowt().await?;
+		let mut cmd = remowt.cmd("fleet-install-secrets");
 		cmd.arg("decrypt").eqarg("--secret", data.to_string());
 		let encoded = cmd
 			.sudo()
@@ -434,8 +381,9 @@
 		data: SecretData,
 		targets: BTreeSet<SecretOwner>,
 	) -> Result<SecretData> {
+		let remowt = self.remowt().await?;
 		ensure!(data.encrypted, "secret is not encrypted");
-		let mut cmd = self.cmd("fleet-install-secrets").await?;
+		let mut cmd = remowt.cmd("fleet-install-secrets");
 		cmd.arg("reencrypt").eqarg("--secret", data.to_string());
 		for target in targets {
 			let key = self.config.key(&target).await?;
@@ -451,74 +399,39 @@
 		Ok(data)
 	}
 	/// Returns path for futureproofing, as path might change i.e on conversion to CA
-	pub async fn remote_derivation(&self, path: &PathBuf) -> Result<PathBuf> {
+	pub async fn remote_derivation(&self, path: impl AsRef<Utf8Path>) -> Result<Utf8PathBuf> {
+		let path = path.as_ref().to_owned();
 		if self.local {
 			// Path is located locally, thus already trusted.
-			return Ok(path.to_owned());
-		}
-		let mut sign = MyCommand::new(
-			// TODO: Look at the current escalation strategy.
-			// ... or switch to run0 right after polkit update
-			EscalationStrategy::Sudo,
-			"nix",
-		);
-		sign.arg("store")
-			.arg("sign")
-			.comparg("--key-file", "/etc/nix/private-key")
-			.arg("-r")
-			.arg(&path);
-		if let Err(e) = sign.sudo().run_nix().await {
-			warn!("failed to sign store paths: {e}");
+			return Ok(path);
 		}
-		let mut nix = MyCommand::new(
-			// Not used
-			EscalationStrategy::Su,
-			"nix",
-		);
-		nix.arg("copy").arg("--substitute-on-destination");
-
-		let proto = if self.legacy_ssh_store.get().cloned().unwrap_or(false) {
-			"ssh"
-		} else {
-			"ssh-ng"
+		let sign: Pin<Box<dyn Future<Output = Result<()>> + Send>> = {
+			let path = path.clone();
+			Box::pin(async move {
+				let local = self.config.local_host();
+				let plugin_id = local.ensure_nix_plugin().await?;
+				let nix = local
+					.remowt()
+					.await?
+					.plugin_endpoints::<remowt_fleet::NixClient<_>>(plugin_id);
+				nix.sign_closure(path, Utf8PathBuf::from("/etc/nix/private-key"))
+					.await
+					.map_err(|e| anyhow!("{e:?}"))?
+					.map_err(|e| anyhow!("{e}"))?;
+				Ok(())
+			})
 		};
-
-		match self.deploy_kind().await? {
-			DeployKind::Fleet | DeployKind::UpgradeToFleet | DeployKind::NixosLustrate => {
-				nix.comparg("--to", format!("{proto}://{}", self.name));
-			}
-			DeployKind::NixosInstall => {
-				nix
-					// Signature checking makes no sense with remote-store store argument set, as we're not even interacting with remote nix daemon
-					.arg("--no-check-sigs")
-					.comparg(
-						"--to",
-						format!("{proto}://root@{}?remote-store=/mnt", self.name),
-					);
-			}
+		if let Err(e) = sign.await {
+			warn!("failed to sign store paths: {e}");
 		}
-		nix.arg(path);
-		nix.run_nix().await.context("nix copy")?;
-		Ok(path.to_owned())
-	}
-	pub async fn systemctl_stop(&self, name: &str) -> Result<()> {
-		let mut cmd = self.cmd("systemctl").await?;
-		cmd.arg("stop").arg(name);
-		cmd.sudo().run().await
-	}
-	pub async fn systemctl_start(&self, name: &str) -> Result<()> {
-		let mut cmd = self.cmd("systemctl").await?;
-		cmd.arg("start").arg(name);
-		cmd.sudo().run().await
-	}
-
-	pub async fn rm_file(&self, path: impl AsRef<OsStr>, sudo: bool) -> Result<()> {
-		let mut cmd = self.cmd("rm").await?;
-		cmd.arg("-f").arg(path);
-		if sudo {
-			cmd = cmd.sudo()
+		let store = self.nix_store().await?;
+		{
+			let path = path.clone();
+			spawn_blocking(move || nix_eval::copy_closure_to(&store, path.as_ref()))
+				.await?
+				.context("copying closure to remote store")?;
 		}
-		cmd.run().await
+		Ok(path)
 	}
 }
 
@@ -630,26 +543,32 @@
 		}
 		Ok(out)
 	}
-	pub fn local_host(&self) -> ConfigHost {
-		ConfigHost {
-			config: self.clone(),
-			name: "<virtual localhost>".to_owned(),
-			host_config: None,
-			nixos_config: OnceLock::new(),
-			nixos_unchecked_config: OnceLock::new(),
-			groups: {
-				let cell = OnceLock::new();
-				let _ = cell.set(vec![]);
-				cell
-			},
-			pkgs_override: Some(self.default_pkgs.clone()),
+	pub fn local_host(&self) -> Arc<ConfigHost> {
+		self.local_host
+			.get_or_init(|| {
+				Arc::new(ConfigHost {
+					config: self.clone(),
+					name: "<virtual localhost>".to_owned(),
+					host_config: None,
+					nixos_config: OnceLock::new(),
+					nixos_unchecked_config: OnceLock::new(),
+					groups: {
+						let cell = OnceLock::new();
+						let _ = cell.set(vec![]);
+						cell
+					},
+					pkgs_override: Some(self.default_pkgs.clone()),
 
-			local: true,
-			session: OnceLock::new(),
-			deploy_kind: OnceLock::new(),
-			session_destination: OnceLock::new(),
-			legacy_ssh_store: OnceLock::new(),
-		}
+					local: true,
+					remowt: OnceLock::new(),
+					nix_store: OnceLock::new(),
+					nix_plugin: tokio::sync::OnceCell::new(),
+					deploy_kind: OnceLock::new(),
+					session_destination: OnceLock::new(),
+					legacy_ssh_store: OnceLock::new(),
+				})
+			})
+			.clone()
 	}
 
 	pub fn preferred_hosts(
@@ -684,7 +603,9 @@
 
 			// TODO: Remove with connectivit refactor
 			local: self.localhost == name,
-			session: OnceLock::new(),
+			remowt: OnceLock::new(),
+			nix_store: OnceLock::new(),
+			nix_plugin: tokio::sync::OnceCell::new(),
 			deploy_kind: OnceLock::new(),
 			session_destination: OnceLock::new(),
 			legacy_ssh_store: OnceLock::new(),
modifiedcrates/fleet-base/src/keys.rsdiffbeforeafterboth
--- a/crates/fleet-base/src/keys.rs
+++ b/crates/fleet-base/src/keys.rs
@@ -31,7 +31,8 @@
 		} else {
 			warn!("Loading key for {}", host);
 			let host = self.host(host)?;
-			let mut cmd = host.cmd("cat").await?;
+			let remowt = host.remowt().await?;
+			let mut cmd = remowt.cmd("cat");
 			cmd.arg("/etc/ssh/ssh_host_ed25519_key.pub");
 			let key = cmd.run_string().await?;
 			self.update_key(&host.name, key.clone());
modifiedcrates/fleet-base/src/lib.rsdiffbeforeafterboth
--- a/crates/fleet-base/src/lib.rs
+++ b/crates/fleet-base/src/lib.rs
@@ -1,8 +1,6 @@
-pub mod command;
 pub mod deploy;
 pub mod fleetdata;
 pub mod host;
 mod keys;
 pub mod opts;
 pub mod primops;
-pub mod secret_storage;
modifiedcrates/fleet-base/src/opts.rsdiffbeforeafterboth
--- a/crates/fleet-base/src/opts.rs
+++ b/crates/fleet-base/src/opts.rs
@@ -1,9 +1,8 @@
 use std::{
 	collections::{BTreeMap, BTreeSet},
 	env::current_dir,
-	ffi::OsString,
 	str::FromStr,
-	sync::{Arc, Mutex},
+	sync::{Arc, OnceLock},
 };
 
 use anyhow::{Context, Result, bail};
@@ -196,7 +195,7 @@
 	}
 
 	// TODO: Config should be detached from opts.
-	pub fn build(&self, nix_args: Vec<OsString>, assert: bool) -> Result<Config> {
+	pub fn build(&self, assert: bool) -> Result<Config> {
 		let cwd = current_dir()?;
 		let mut directory = cwd.clone();
 		let mut fleet_data_path = directory.join("fleet.nix");
@@ -274,11 +273,11 @@
 			data,
 			flake_outputs: flake,
 			local_system: self.local_system.clone(),
-			nix_args,
 			config_field,
 			default_pkgs,
 			nixpkgs,
 			localhost: self.localhost.to_owned(),
+			local_host: OnceLock::new(),
 		}));
 
 		PRIMOPS_DATA
modifiedcrates/fleet-base/src/primops.rsdiffbeforeafterboth
--- a/crates/fleet-base/src/primops.rs
+++ b/crates/fleet-base/src/primops.rs
@@ -1,10 +1,12 @@
 use std::collections::{BTreeMap, BTreeSet, HashMap};
-use std::sync::OnceLock;
+use std::sync::{Arc, OnceLock};
 
 use anyhow::{Context, bail, ensure};
 use fleet_shared::SecretData;
 use itertools::Itertools;
 use nix_eval::{NativeFn, Value, await_in_nix, nix_go, nix_go_json};
+use remowt_endpoints::fs::FsClient;
+use remowt_link_shared::BifConfig;
 use serde::Deserialize;
 use tracing::{info, warn};
 
@@ -28,7 +30,7 @@
 	let pkgs = host_on.pkgs()?;
 	let default_mk_secret_generators = nix_go!(pkgs.mkSecretGenerators);
 	let generators = nix_go!(default_mk_secret_generators(Obj { recipients }));
-	Ok(pkgs.clone().attrs_update(generators)?)
+	pkgs.clone().attrs_update(generators)
 }
 pub fn get_default_pkgs_and_generators(config: &Config) -> Result<Value> {
 	let host_on = config.local_host();
@@ -91,12 +93,18 @@
 			let impure_on: Option<String> = nix_go_json!(default_generator_drv.impureOn);
 
 			let host_on = if let Some(on) = &impure_on {
-				config
-					.host(on)
-					.context("failed to get secret generation target host")?
+				Arc::new(
+					config
+						.host(on)
+						.context("failed to get secret generation target host")?,
+				)
 			} else {
 				config.local_host()
 			};
+
+			let remowt = host_on.remowt().await?;
+			let fs = remowt.endpoints::<FsClient<BifConfig>>();
+
 			let mut recipients = Vec::new();
 			for owner in &expectations.owners {
 				recipients.push(config.key(owner).await?);
@@ -116,12 +124,13 @@
 				.context("failed to copy generator to target host")?;
 
 			// TODO: Remove destdir after everything is done
-			let out_parent = host_on
+			let out_parent = fs
 				.mktemp_dir()
 				.await
+				.map_err(|e| anyhow!("{e:?}"))
 				.context("failed to prepare generator output dir on target host")?;
-			let out = format!("{out_parent}/out");
-			let mut generator_cmd = host_on.cmd(generator).await?;
+			let out = out_parent.join("out");
+			let mut generator_cmd = remowt.cmd(generator);
 			generator_cmd.env("out", &out);
 			if impure_on.is_none() {
 				let project_path: String = config
@@ -138,7 +147,7 @@
 				.context("failed to run impure generator")?;
 
 			{
-				let marker = host_on.read_file_text(format!("{out}/marker")).await?;
+				let marker = fs.read_file_text(out.join("marker")).await?;
 				ensure!(
 					marker == "SUCCESS",
 					"impure generator ended prematurely, secret generation failed"
@@ -147,15 +156,16 @@
 
 			let mut missing_parts = expectations.parts.clone();
 			let mut parts = BTreeMap::new();
-			for part in host_on.read_dir(&out).await? {
+			for part in fs.read_dir(&out).await? {
+				let part = part.into_string();
 				if part == "created_at" || part == "expires_at" || part == "marker" {
 					continue;
 				}
 				let Some(part_def) = missing_parts.remove(&part) else {
 					bail!("secret generator has produced an unexpected part: {part}");
 				};
-				let contents: SecretData = host_on
-					.read_file_text(format!("{out}/{part}"))
+				let contents: SecretData = fs
+					.read_file_text(out.join("part"))
 					.await?
 					.parse()
 					.map_err(|e| anyhow!("failed to decode secret {out:?} part {part:?}: {e}"))?;
@@ -178,11 +188,12 @@
 				);
 			}
 
-			let created_at = host_on.read_file_value(format!("{out}/created_at")).await?;
-			let expires_at = host_on
-				.read_file_value(format!("{out}/expires_at"))
-				.await
-				.ok();
+			let created_at = fs.read_file_value(out.join("created_at")).await??;
+			let expires_at = match fs.read_file_value(out.join("expires_at")).await {
+				Ok(v) => Some(v?),
+				Err(remowt_endpoints::fs::Error::NotFound) => None,
+				Err(e) => return Err(e.into()),
+			};
 
 			let new_data = FleetSecretData {
 				created_at,
deletedcrates/fleet-base/src/secret_storage.rsdiffbeforeafterboth
--- a/crates/fleet-base/src/secret_storage.rs
+++ /dev/null
@@ -1,278 +0,0 @@
-use anyhow::{Result, bail, ensure};
-use itertools::Itertools;
-use std::fs::{File, metadata};
-use std::io::{self, ErrorKind, Read, Write};
-use std::path::PathBuf;
-use std::str::FromStr;
-use std::{env, fs};
-
-use tempfile::{TempPath, tempfile_in};
-use toml_edit::{Document, DocumentMut, Formatted, Item, Value};
-
-struct Name(String);
-
-fn encode_name(name: &str) -> Name {
-	assert!(
-		!name.starts_with(['_', '.']),
-		"groups should not start with _ or ."
-	);
-	assert!(
-		!name.chars().any(|c| c == '/'),
-		"group name should not contain internal slash"
-	);
-	Name(name.to_owned())
-}
-
-enum RewriteError {
-	ConcurrentCreate,
-	ConcurrentDelete,
-	ConcurrentModify,
-	ConcurrentWrite,
-	Io(io::Error),
-	Persist(tempfile::PersistError),
-}
-
-fn safe_rewrite(
-	path: &PathBuf,
-	old_content: Option<Vec<u8>>,
-	new_content: Option<Vec<u8>>,
-) -> Result<(), RewriteError> {
-	let mut f = match (old_content.is_some(), new_content.is_some()) {
-		(false, true) => match File::create_new(path) {
-			Ok(v) => v,
-			Err(e) if e.kind() == ErrorKind::AlreadyExists => {
-				return Err(RewriteError::ConcurrentCreate);
-			}
-			Err(e) => return Err(RewriteError::Io(e)),
-		},
-		(true, _) => match File::open(&path) {
-			Ok(v) => v,
-			Err(_e) => return Err(RewriteError::ConcurrentDelete),
-		},
-		(false, false) => match metadata(&path) {
-			Err(e) if e.kind() == ErrorKind::NotFound => {
-				return Ok(());
-			}
-			Ok(_) => return Err(RewriteError::ConcurrentCreate),
-			Err(e) => return Err(RewriteError::Io(e)),
-		},
-	};
-	f.lock().map_err(RewriteError::Io)?;
-	let mut check_content = vec![];
-	f.read_to_end(&mut check_content)
-		.map_err(RewriteError::Io)?;
-	match &old_content {
-		Some(old) => {
-			if old != &check_content {
-				return Err(RewriteError::ConcurrentModify);
-			}
-		}
-		None => {
-			if !check_content.is_empty() {
-				return Err(RewriteError::ConcurrentDelete);
-			}
-		}
-	}
-	if let Some(new_content) = new_content {
-		if Some(&new_content) == old_content.as_ref() {
-			return Ok(());
-		}
-		let dir = path.parent().expect("file is in directory, thus not root");
-		let mut tempfile = tempfile::Builder::new()
-			.prefix(".rewrite-")
-			.tempfile_in(dir)
-			.map_err(RewriteError::Io)?;
-		tempfile.write_all(&new_content).map_err(RewriteError::Io)?;
-		tempfile.flush().map_err(RewriteError::Io)?;
-		tempfile.persist(path).map_err(RewriteError::Persist)?;
-	} else {
-		fs::remove_file(path).map_err(RewriteError::Io)?;
-	}
-	let _ = f.unlock();
-	Ok(())
-}
-fn update_string(path: PathBuf, modify: impl Fn(&mut Option<String>) -> Result<()>) -> Result<()> {
-	loop {
-		let orig = match fs::read_to_string(&path) {
-			Ok(v) => Some(v),
-			Err(e) if e.kind() == ErrorKind::NotFound => None,
-			Err(e) => return Err(e.into()),
-		};
-		let mut edit = orig.clone();
-		modify(&mut edit);
-
-		match safe_rewrite(&path, orig.map(String::into), edit.map(String::into)) {
-			Ok(()) => return Ok(()),
-			Err(
-				RewriteError::ConcurrentCreate
-				| RewriteError::ConcurrentModify
-				| RewriteError::ConcurrentWrite
-				| RewriteError::ConcurrentDelete,
-			) => {
-				continue;
-			}
-			Err(RewriteError::Io(io)) => return Err(io.into()),
-			Err(RewriteError::Persist(io)) => return Err(io.into()),
-		}
-	}
-}
-fn update_toml(path: PathBuf, modify: impl Fn(&mut DocumentMut) -> Result<()>) -> Result<()> {
-	update_string(path, |str| {
-		let mut doc = match str {
-			None => DocumentMut::new(),
-			Some(v) => DocumentMut::from_str(v)?,
-		};
-		modify(&mut doc)?;
-		if doc.is_empty() {
-			*str = None
-		} else {
-			*str = Some(doc.to_string())
-		}
-		Ok(())
-	})
-}
-fn update_lines(path: PathBuf, modify: impl Fn(&mut Vec<String>) -> Result<()>) -> Result<()> {
-	update_string(path, |str| {
-		let mut list = if let Some(str) = str {
-			str.split('\n').map(|s| s.to_owned()).collect_vec()
-		} else {
-			vec![]
-		};
-		let had_end_newline = if list.last().map(|v| v.as_str()) == Some("") {
-			list.pop();
-			true
-		} else {
-			false
-		};
-		modify(&mut list)?;
-		if list.is_empty() {
-			*str = None
-		} else {
-			if had_end_newline {
-				list.push("".to_owned())
-			}
-			*str = Some(list.join("\n"));
-		}
-		Ok(())
-	})
-}
-fn update_section(
-	data: &mut Vec<String>,
-	start: &str,
-	end: &str,
-	modify: impl Fn(&mut Vec<String>) -> Result<()>,
-) -> Result<()> {
-	let first = data
-		.iter()
-		.enumerate()
-		.filter(|(_, v)| *v == start)
-		.at_most_one()
-		.map_err(|_| anyhow::anyhow!("there should be at most one section start"))?
-		.map(|(v, _)| v);
-	let last = data
-		.iter()
-		.enumerate()
-		.filter(|(_, v)| *v == end)
-		.at_most_one()
-		.map_err(|_| anyhow::anyhow!("there should be at most one section end"))?
-		.map(|(v, _)| v);
-
-	match (first, last) {
-		(None, None) => {
-			let mut out = Vec::new();
-			modify(&mut out)?;
-			if out.is_empty() {
-				return Ok(());
-			}
-			data.push(start.to_owned());
-			data.extend(out);
-			data.push(end.to_owned());
-			Ok(())
-		}
-		(None, Some(_)) | (Some(_), None) => {
-			bail!("mismatched section start/end")
-		}
-		(Some(first), Some(last)) => {
-			ensure!(first < last, "section end should come after start");
-			let mut out = data[first + 1..last]
-				.iter()
-				.map(|v| v.to_owned())
-				.collect_vec();
-			modify(&mut out)?;
-			if out.is_empty() {
-				data.drain(first..=last);
-			} else {
-				data.splice(first + 1..last, out);
-			}
-			Ok(())
-		}
-	}
-}
-
-struct Group {
-	path: PathBuf,
-}
-impl Group {
-	fn new(path: PathBuf) -> Self {
-		Self { path }
-	}
-	fn manage(&self, manager: &str) {}
-	fn ensure_managing(&self, manager: &str) {
-		if !self.has_stored() {
-			return;
-		}
-		let managed = match fs::read_to_string(self.path.join(".managed_by")) {
-			Ok(found_manager) => found_manager.lines().any(|line| line == manager),
-			Err(e) if e.kind() == ErrorKind::NotFound => true,
-			Err(e) => panic!("{e}"),
-		};
-		assert!(managed);
-	}
-	fn has_stored(&self) -> bool {
-		match fs::metadata(&self.path) {
-			Ok(d) => d.is_dir(),
-			Err(e) if e.kind() == ErrorKind::NotFound => false,
-			Err(e) => panic!("{e}"),
-		}
-	}
-}
-
-struct Root {
-	path: PathBuf,
-}
-impl Root {
-	fn new(path: PathBuf) -> Self {
-		Self { path }
-	}
-	fn subgroup(&self, name: &str) -> Group {
-		Group::new(self.path.join(name))
-	}
-}
-
-#[test]
-fn test() {
-	let mut data = vec![
-		"a".to_owned(),
-		"b".to_owned(),
-		"start".to_owned(),
-		"c".to_owned(),
-		"d".to_owned(),
-		"end".to_owned(),
-		"e".to_owned(),
-		"f".to_owned(),
-	];
-	update_section(&mut data, "start", "end", |a| {
-		a.push("vv".to_owned());
-		Ok(())
-	})
-	.unwrap();
-	dbg!(&data);
-	// for v in 0..1000 {
-	// 	update_toml(PathBuf::from("./test.toml"), |e| {
-	// 		e.as_table_mut()
-	// 			.insert("hello", Item::Value(Value::Integer(Formatted::new(v))));
-	// 	})
-	// 	.expect("update")
-	// }
-	// v.subgroup(name)
-}
modifiedcrates/nix-eval/Cargo.tomldiffbeforeafterboth
--- a/crates/nix-eval/Cargo.toml
+++ b/crates/nix-eval/Cargo.toml
@@ -10,10 +10,11 @@
 nixlike.workspace = true
 serde = { workspace = true, features = ["derive"] }
 serde_json.workspace = true
-thiserror.workspace = true
 tracing.workspace = true
 
+camino = { workspace = true, features = ["serde1"] }
 cxx.workspace = true
+futures.workspace = true
 itertools.workspace = true
 test-log.workspace = true
 tokio.workspace = true
modifiedcrates/nix-eval/src/drv.rsdiffbeforeafterboth
--- a/crates/nix-eval/src/drv.rs
+++ b/crates/nix-eval/src/drv.rs
@@ -66,7 +66,13 @@
 #[derive(Debug, Deserialize)]
 pub struct DrvParsed {
 	pub inputs: DrvInputs,
-	pub outputs: HashMap<String, serde_json::Value>,
+	pub outputs: HashMap<String, DrvParsedOutput>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct DrvParsedOutput {
+	#[serde(default)]
+	pub path: Option<String>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -82,18 +88,19 @@
 	pub outputs: Vec<String>,
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct DrvGraph {
 	pub root: String,
 	pub nodes: HashMap<String, DrvNode>,
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct DrvNode {
 	pub name: String,
 	pub input_drvs: HashMap<String, Vec<String>>,
 	pub input_srcs: Vec<String>,
-	pub outputs: Vec<String>,
+	// TODO: CA outputs without a known paths are skipped
+	pub outputs: HashMap<String, String>,
 }
 
 impl DrvGraph {
@@ -124,19 +131,59 @@
 				}
 			}
 
+			let outputs: HashMap<String, String> = parsed
+				.outputs
+				.into_iter()
+				.filter_map(|(name, out)| out.path.map(|p| (name, to_absolute_store_path(&sd, &p))))
+				.collect();
+
 			nodes.insert(
 				path.clone(),
 				DrvNode {
 					name: extract_drv_name(&path),
 					input_drvs,
 					input_srcs: parsed.inputs.srcs,
-					outputs: parsed.outputs.into_keys().collect(),
+					outputs,
 				},
 			);
 		}
 
 		Ok(Self { root, nodes })
 	}
+
+	pub fn wanted_outputs(&self, root_outputs: &[String]) -> HashMap<String, Vec<String>> {
+		let mut wanted: HashMap<String, HashSet<String>> = HashMap::new();
+		wanted.insert(self.root.clone(), root_outputs.iter().cloned().collect());
+
+		let mut queue: VecDeque<String> = VecDeque::new();
+		queue.push_back(self.root.clone());
+		while let Some(path) = queue.pop_front() {
+			let Some(node) = self.nodes.get(&path) else {
+				continue;
+			};
+			for (dep_path, dep_outputs) in &node.input_drvs {
+				let entry = wanted.entry(dep_path.clone()).or_default();
+				let mut changed = false;
+				for o in dep_outputs {
+					if entry.insert(o.clone()) {
+						changed = true;
+					}
+				}
+				if changed {
+					queue.push_back(dep_path.clone());
+				}
+			}
+		}
+
+		wanted
+			.into_iter()
+			.map(|(k, v)| {
+				let mut v: Vec<_> = v.into_iter().collect();
+				v.sort();
+				(k, v)
+			})
+			.collect()
+	}
 }
 
 fn extract_drv_name(drv_path: &str) -> String {
modifiedcrates/nix-eval/src/lib.ccdiffbeforeafterboth
--- a/crates/nix-eval/src/lib.cc
+++ b/crates/nix-eval/src/lib.cc
@@ -1,8 +1,25 @@
 #include "nix-eval/src/lib.rs"
 #include "lib.hh"
 #include <nix/fetchers/fetch-settings.hh>
+#include <nix/store/build-result.hh>
+#include <nix/store/content-address.hh>
+#include <nix/store/derived-path.hh>
+#include <nix/store/local-fs-store.hh>
+#include <nix/store/outputs-spec.hh>
+#include <nix/store/path-info.hh>
+#include <nix/store/profiles.hh>
+#include <nix/store/realisation.hh>
+#include <nix/store/store-api.hh>
+#include <nix/util/file-system.hh>
+#include <nix/util/hash.hh>
+#include <nix/util/posix-source-accessor.hh>
 #include <nix/util/ref.hh>
+#include <nix/util/signature/local-keys.hh>
+#include <nix/util/signature/signer.hh>
+#include <nix/util/source-path.hh>
 #include <nix_api_fetchers.h>
+#include <nix_api_store_internal.h>
+#include <sstream>
 
 struct nix_fetchers_settings {
   nix::ref<nix::fetchers::Settings> settings;
@@ -15,3 +32,175 @@
   bool result = settings_ref->set(setting, value);
 }
 }
+
+rust::String switch_profile(Store *store, rust::Str profile,
+                            rust::Str store_path) {
+  try {
+    auto nixStore = store->ptr;
+    auto *lfs = dynamic_cast<nix::LocalFSStore *>(&*nixStore);
+    if (!lfs)
+      return rust::String("destination is not a local-fs store");
+    auto path = nixStore->parseStorePath(std::string(store_path));
+    std::filesystem::path prof = std::string(profile);
+    auto gen = nix::createGeneration(*lfs, prof, path);
+    nix::switchLink(prof, gen);
+    return rust::String();
+  } catch (const std::exception &e) {
+    return rust::String(e.what());
+  }
+}
+
+rust::String sign_closure(Store *store, rust::Str store_path,
+                          rust::Str key_file) {
+  try {
+    auto nixStore = store->ptr;
+    nix::LocalSigner signer(
+        nix::SecretKey::parse(nix::readFile(std::string(key_file))));
+    auto root = nixStore->parseStorePath(std::string(store_path));
+    nix::StorePathSet closure;
+    nixStore->computeFSClosure(root, closure);
+    for (auto &p : closure) {
+      auto info = nixStore->queryPathInfo(p);
+      nix::ValidPathInfo info2(*info);
+      info2.sign(*nixStore, signer);
+      nixStore->addSignatures(p, info2.sigs);
+    }
+    return rust::String();
+  } catch (const std::exception &e) {
+    return rust::String(e.what());
+  }
+}
+
+CxxListGenerationsResult list_generations(rust::Str profile_path) {
+  CxxListGenerationsResult out{rust::String(), {}};
+  try {
+    auto [gens, current] =
+        nix::findGenerations(std::filesystem::path(std::string(profile_path)));
+    for (auto &g : gens) {
+      CxxProfileGeneration cg{};
+      cg.id = g.number;
+      cg.store_path = rust::String(g.path.string());
+      cg.creation_time_unix = static_cast<int64_t>(g.creationTime);
+      cg.current = current.has_value() && *current == g.number;
+      out.generations.push_back(cg);
+    }
+  } catch (const std::exception &e) {
+    out.error = rust::String(e.what());
+  }
+  return out;
+}
+
+static std::vector<std::string> split_lines(rust::Str joined) {
+  std::vector<std::string> out;
+  std::string buf;
+  std::istringstream iss((std::string(joined)));
+  while (std::getline(iss, buf)) {
+    if (!buf.empty())
+      out.push_back(buf);
+  }
+  return out;
+}
+
+CxxBuildResult build_drv_outputs(Store *store, rust::Str drv_path,
+                                 // TODO: Vec<str>
+                                 rust::Str output_names_joined) {
+  CxxBuildResult res{rust::String(), {}};
+  try {
+    auto nixStore = store->ptr;
+    auto sp = nixStore->parseStorePath(std::string(drv_path));
+
+    nix::DerivedPath::Built built{
+        .drvPath = nix::makeConstantStorePathRef(sp),
+        .outputs = nix::OutputsSpec{nix::OutputsSpec::All{}},
+    };
+    auto names = split_lines(output_names_joined);
+    if (!names.empty()) {
+      std::set<nix::OutputName, std::less<>> nameSet;
+      for (auto &n : names)
+        nameSet.insert(n);
+      built.outputs =
+          nix::OutputsSpec{nix::OutputsSpec::Names{std::move(nameSet)}};
+    }
+
+    std::vector<nix::DerivedPath> reqs;
+    reqs.push_back(built);
+
+    auto results = nixStore->buildPathsWithResults(reqs);
+    for (auto &r : results) {
+      if (auto *failure = r.tryGetFailure()) {
+        res.error = rust::String(failure->what());
+        return res;
+      }
+    }
+    for (auto &r : results) {
+      if (auto *success = r.tryGetSuccess()) {
+        for (auto &[name, real] : success->builtOutputs) {
+          res.outputs.push_back(
+              rust::String(nixStore->printStorePath(real.outPath)));
+        }
+      }
+    }
+  } catch (const std::exception &e) {
+    res.error = rust::String(e.what());
+  }
+  return res;
+}
+
+CxxBuildResult substitute_paths(Store *store, rust::Str paths_joined) {
+  CxxBuildResult res{rust::String(), {}};
+  try {
+    auto nixStore = store->ptr;
+    auto paths = split_lines(paths_joined);
+
+    std::vector<nix::DerivedPath> reqs;
+    reqs.reserve(paths.size());
+    for (auto &p : paths) {
+      reqs.push_back(nix::DerivedPath::Opaque{nixStore->parseStorePath(p)});
+    }
+
+    // Substituter miss will cause scheduler to trigger build
+    auto results = nixStore->buildPathsWithResults(reqs);
+    for (auto &r : results) {
+      if (r.tryGetSuccess() == nullptr)
+        continue;
+      if (auto *opaque = std::get_if<nix::DerivedPath::Opaque>(&r.path.raw())) {
+        res.outputs.push_back(
+            rust::String(nixStore->printStorePath(opaque->path)));
+      }
+    }
+  } catch (const std::exception &e) {
+    res.error = rust::String(e.what());
+  }
+  return res;
+}
+
+bool is_valid_path(Store *store, rust::Str path) {
+  try {
+    auto nixStore = store->ptr;
+    auto sp = nixStore->parseStorePath(std::string(path));
+    return nixStore->isValidPath(sp);
+  } catch (const std::exception &) {
+    return false;
+  }
+}
+
+AddFileToStoreResult add_file_to_store(Store *store, rust::Str name,
+                                       rust::Str path) {
+  AddFileToStoreResult out{rust::String(), rust::String(), rust::String()};
+  try {
+    auto nixStore = store->ptr;
+    auto src = nix::PosixSourceAccessor::createAtRoot(
+        std::filesystem::path(std::string(path)));
+    auto info = nixStore->addToStoreSlow(std::string(name), src,
+                                         nix::ContentAddressMethod::Raw::Flat,
+                                         nix::HashAlgorithm::SHA256);
+    out.store_path = rust::String(nixStore->printStorePath(info.path));
+    if (info.ca.has_value()) {
+      out.hash =
+          rust::String(info.ca->hash.to_string(nix::HashFormat::SRI, true));
+    }
+  } catch (const std::exception &e) {
+    out.error = rust::String(e.what());
+  }
+  return out;
+}
modifiedcrates/nix-eval/src/lib.hhdiffbeforeafterboth
--- a/crates/nix-eval/src/lib.hh
+++ b/crates/nix-eval/src/lib.hh
@@ -1,7 +1,32 @@
 #pragma once
+#include "rust/cxx.h"
 #include <nix_api_fetchers.h>
 
+struct Store;
+
+struct AddFileToStoreResult;
+struct CxxBuildResult;
+
 extern "C" {
 void set_fetcher_setting(nix_fetchers_settings *settings, const char *setting,
                          const char *value);
 }
+
+rust::String switch_profile(Store *store, rust::Str profile,
+                            rust::Str store_path);
+
+rust::String sign_closure(Store *store, rust::Str store_path,
+                          rust::Str key_file);
+
+struct CxxListGenerationsResult;
+CxxListGenerationsResult list_generations(rust::Str profile_path);
+
+AddFileToStoreResult add_file_to_store(Store *store, rust::Str name,
+                                       rust::Str path);
+
+CxxBuildResult build_drv_outputs(Store *store, rust::Str drv_path,
+                                 rust::Str output_names_joined);
+
+CxxBuildResult substitute_paths(Store *store, rust::Str paths_joined);
+
+bool is_valid_path(Store *store, rust::Str path);
modifiedcrates/nix-eval/src/lib.rsdiffbeforeafterboth
--- a/crates/nix-eval/src/lib.rs
+++ b/crates/nix-eval/src/lib.rs
@@ -1,12 +1,13 @@
 use std::borrow::Cow;
 use std::cell::RefCell;
+use std::collections::HashMap;
 use std::ffi::{CStr, CString, c_char, c_int, c_uint, c_void};
 use std::ptr::{null, null_mut};
 use std::sync::{Arc, LazyLock, OnceLock};
 use std::{array, fmt, slice};
-use std::{collections::HashMap, path::PathBuf};
 
 use anyhow::{Context, anyhow, bail};
+use camino::{Utf8Path, Utf8PathBuf};
 use itertools::Itertools;
 use serde::Serialize;
 use serde::de::DeserializeOwned;
@@ -38,8 +39,9 @@
 	make_attrs, make_bindings_builder, make_list, make_list_builder, realised_string,
 	realised_string_free, realised_string_get_buffer_size, realised_string_get_buffer_start,
 	realised_string_get_store_path, realised_string_get_store_path_count, register_primop,
-	set_err_msg, setting_set, state_free, store_open, store_parse_path, store_path_free,
-	store_path_name, string_realise, value, value_call, value_decref, value_force, value_incref,
+	set_err_msg, setting_set, state_free, store_copy_closure, store_free, store_open,
+	store_parse_path, store_path_free, store_path_name, string_realise, value, value_call,
+	value_decref, value_force, value_incref,
 };
 
 // Contains macros helpers
@@ -47,6 +49,7 @@
 pub mod logging;
 #[doc(hidden)]
 pub mod macros;
+pub mod scheduler;
 
 #[doc(hidden)]
 pub mod __macro_support {
@@ -68,8 +71,28 @@
 }
 #[cxx::bridge]
 pub mod nix_cxx {
+	struct AddFileToStoreResult {
+		error: String,
+		store_path: String,
+		hash: String,
+	}
+	struct CxxProfileGeneration {
+		id: u64,
+		store_path: String,
+		creation_time_unix: i64,
+		current: bool,
+	}
+	struct CxxListGenerationsResult {
+		error: String,
+		generations: Vec<CxxProfileGeneration>,
+	}
+	struct CxxBuildResult {
+		error: String,
+		outputs: Vec<String>,
+	}
 	unsafe extern "C++" {
 		type nix_fetchers_settings;
+		type Store;
 		include!("nix-eval/src/lib.hh");
 
 		#[allow(clippy::missing_safety_doc)]
@@ -78,6 +101,34 @@
 			setting: *const c_char,
 			value: *const c_char,
 		);
+
+		#[allow(clippy::missing_safety_doc)]
+		unsafe fn switch_profile(store: *mut Store, profile: &str, store_path: &str) -> String;
+
+		#[allow(clippy::missing_safety_doc)]
+		unsafe fn sign_closure(store: *mut Store, store_path: &str, key_file: &str) -> String;
+
+		#[allow(clippy::missing_safety_doc)]
+		unsafe fn add_file_to_store(
+			store: *mut Store,
+			name: &str,
+			path: &str,
+		) -> AddFileToStoreResult;
+
+		fn list_generations(profile_path: &str) -> CxxListGenerationsResult;
+
+		#[allow(clippy::missing_safety_doc)]
+		unsafe fn build_drv_outputs(
+			store: *mut Store,
+			drv_path: &str,
+			output_names_joined: &str,
+		) -> CxxBuildResult;
+
+		#[allow(clippy::missing_safety_doc)]
+		unsafe fn substitute_paths(store: *mut Store, paths_joined: &str) -> CxxBuildResult;
+
+		#[allow(clippy::missing_safety_doc)]
+		unsafe fn is_valid_path(store: *mut Store, path: &str) -> bool;
 	}
 }
 
@@ -186,7 +237,7 @@
 		unsafe { set_err_msg(self.0, err as c_int, msg.as_ptr()) };
 	}
 	pub fn set_err(&mut self, err: anyhow::Error) {
-		let mut fmt = format!("{err:?}").replace("\0", "\\0");
+		let fmt = format!("{err:?}").replace("\0", "\\0");
 		self.set_err_raw(
 			NixErrorKind::Generic,
 			&CString::new(fmt).expect("NUL bytes were just replaced"),
@@ -352,6 +403,125 @@
 	with_default_context(|c, _| unsafe { setting_set(c, s.as_ptr(), v.as_ptr()) }).map(|_| ())
 }
 
+#[instrument(skip(dst))]
+pub fn copy_closure_to(dst: &Store, path: &Utf8Path) -> Result<()> {
+	let path_c = CString::new(path.as_str())?;
+	with_store_context(|c, src_store, _state| -> Result<()> {
+		let sp = unsafe { store_parse_path(c, src_store, path_c.as_ptr()) };
+		if sp.is_null() {
+			bail!("failed to parse store path {path}");
+		}
+		let rc = unsafe { store_copy_closure(c, src_store, dst.0, sp) };
+		unsafe { store_path_free(sp) };
+		if rc != nix_raw::err_NIX_OK {
+			bail!("store_copy_closure failed (code {rc})");
+		}
+		Ok(())
+	})?
+}
+
+#[instrument]
+pub fn switch_profile(profile: &str, store_path: &Utf8Path) -> Result<()> {
+	let msg = with_store_context(|_c, store, _state| unsafe {
+		nix_cxx::switch_profile(store.cast(), profile, store_path.as_str())
+	})?
+	.to_string();
+	if msg.is_empty() {
+		Ok(())
+	} else {
+		bail!("failed to switch profile {profile}: {msg}");
+	}
+}
+
+// TODO: fleet operator-managed key file
+#[instrument]
+pub fn sign_closure(store_path: &str, key_file: &str) -> Result<()> {
+	let msg = with_store_context(|_c, store, _state| unsafe {
+		nix_cxx::sign_closure(store.cast(), store_path, key_file)
+	})?
+	.to_string();
+	if msg.is_empty() {
+		Ok(())
+	} else {
+		bail!("failed to sign {store_path}: {msg}");
+	}
+}
+
+#[derive(Debug)]
+pub struct AddedFile {
+	pub store_path: Utf8PathBuf,
+	pub hash: String,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct ProfileGeneration {
+	pub id: u64,
+	pub store_path: Utf8PathBuf,
+	pub creation_time_unix: i64,
+	pub current: bool,
+}
+
+#[instrument]
+pub fn list_generations(profile_path: &str) -> Result<Vec<ProfileGeneration>> {
+	let res = nix_cxx::list_generations(profile_path);
+	if !res.error.is_empty() {
+		bail!(
+			"failed to list generations at {profile_path}: {}",
+			res.error
+		);
+	}
+	Ok(res
+		.generations
+		.into_iter()
+		.map(|g| ProfileGeneration {
+			id: g.id,
+			store_path: Utf8PathBuf::from(g.store_path),
+			creation_time_unix: g.creation_time_unix,
+			current: g.current,
+		})
+		.collect())
+}
+
+#[instrument]
+pub fn add_file_to_store(name: &str, path: &Utf8Path) -> Result<AddedFile> {
+	let res = with_store_context(|_c, store, _state| unsafe {
+		nix_cxx::add_file_to_store(store.cast(), name, path.as_str())
+	})?;
+	if !res.error.is_empty() {
+		bail!("failed to add {path} to store: {}", res.error);
+	}
+	Ok(AddedFile {
+		store_path: Utf8PathBuf::from(res.store_path),
+		hash: res.hash,
+	})
+}
+
+pub fn build_drv_outputs(drv_path: &str, output_names: &[String]) -> Result<Vec<String>> {
+	let joined = output_names.join("\n");
+	let res = with_store_context(|_c, store, _state| unsafe {
+		nix_cxx::build_drv_outputs(store.cast(), drv_path, &joined)
+	})?;
+	if !res.error.is_empty() {
+		bail!("build of {drv_path} failed: {}", res.error);
+	}
+	Ok(res.outputs)
+}
+
+pub fn substitute_paths(paths: &[String]) -> Result<Vec<String>> {
+	let joined = paths.join("\n");
+	let res = with_store_context(|_c, store, _state| unsafe {
+		nix_cxx::substitute_paths(store.cast(), &joined)
+	})?;
+	if !res.error.is_empty() {
+		warn!("substitute_paths reported: {}", res.error);
+	}
+	Ok(res.outputs)
+}
+
+pub fn is_valid_path(path: &str) -> Result<bool> {
+	with_store_context(|_c, store, _state| unsafe { nix_cxx::is_valid_path(store.cast(), path) })
+}
+
 pub struct FetchSettings(*mut fetchers_settings);
 impl FetchSettings {
 	pub fn new() -> Self {
@@ -450,17 +620,31 @@
 	unsafe { *user_data.cast::<String>() = s.to_owned() };
 }
 
-struct Store(*mut c_store);
+pub struct Store(*mut c_store);
 unsafe impl Send for Store {}
 unsafe impl Sync for Store {}
 
 impl Store {
+	pub fn open(uri: &str) -> Result<Self> {
+		let uri = CString::new(uri)?;
+		let ptr = with_default_context(|c, _| unsafe { store_open(c, uri.as_ptr(), null_mut()) })?;
+		if ptr.is_null() {
+			bail!("failed to open store");
+		}
+		Ok(Store(ptr))
+	}
+
 	fn parse_path(&self, path: &CStr) -> Result<StorePath> {
 		with_default_context(|c, _| {
 			StorePath(unsafe { store_parse_path(c, self.0, path.as_ptr()) })
 		})
 	}
 }
+impl Drop for Store {
+	fn drop(&mut self) {
+		unsafe { store_free(self.0) }
+	}
+}
 
 #[repr(transparent)]
 pub struct EvalState(*mut c_eval_state);
@@ -858,7 +1042,7 @@
 		Ok(out)
 	}
 	#[instrument(name = "build", skip(self), fields(output))]
-	pub fn build(&self, output: &str) -> Result<PathBuf> {
+	pub fn build(&self, output: &str) -> Result<Utf8PathBuf> {
 		if !self.is_derivation() {
 			bail!("expected derivation to build")
 		}
@@ -880,14 +1064,15 @@
 			.get_field("drvPath")
 			.context("getting drvPath")?
 			.to_string()?;
-		let graph = drv::DrvGraph::resolve(&drv_path)?;
+		let graph = Arc::new(drv::DrvGraph::resolve(&drv_path)?);
 		let _guard = logging::register_build_graph(&Span::current(), &graph);
 
-		// to_string here blocks until the path is built
+		scheduler::build_graph_sync(graph.clone(), vec![output.to_owned()])?;
+
 		let s = v.builtin_to_string()?;
 		let rs = s.to_realised_string()?;
 		let out_path = rs.as_str().to_owned();
-		Ok(PathBuf::from(out_path))
+		Ok(Utf8PathBuf::from(out_path))
 	}
 	pub fn as_json<T: DeserializeOwned>(&self) -> Result<T> {
 		let to_json = Self::eval("builtins.toJSON")?;
modifiedcrates/nix-eval/src/logging.ccdiffbeforeafterboth
--- a/crates/nix-eval/src/logging.cc
+++ b/crates/nix-eval/src/logging.cc
@@ -96,7 +96,9 @@
 
 extern "C" {
 void apply_tracing_logger() {
-  logger = std::make_unique<TracingLogger>();
+  auto old = logger;
+  logger = new TracingLogger();
+  delete old;
   // verbosity = lvlVomit;
 }
 rust::Box<ErrorInfoBuilder>
modifiedcrates/nix-eval/src/logging.rsdiffbeforeafterboth
--- a/crates/nix-eval/src/logging.rs
+++ b/crates/nix-eval/src/logging.rs
@@ -499,7 +499,9 @@
 			} else if level == Level::INFO {
 				info!(target: "nix", "{}", s)
 			} else if level == Level::DEBUG {
-				debug!(target: "nix", "{}", s)
+				if s != "querying info about missing paths" {
+					debug!(target: "nix", "{}", s)
+				}
 			} else {
 				trace!(target: "nix", "{}", s)
 			}
@@ -583,7 +585,9 @@
 	} else if level == Level::INFO {
 		info!(target: "nix", "{v}")
 	} else if level == Level::DEBUG {
-		debug!(target: "nix", "{v}")
+		if v != "querying info about missing paths" {
+			debug!(target: "nix", "{v}")
+		}
 	} else {
 		trace!(target: "nix", "{v}")
 	}
addedcrates/nix-eval/src/scheduler.rsdiffbeforeafterboth
--- /dev/null
+++ b/crates/nix-eval/src/scheduler.rs
@@ -0,0 +1,372 @@
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+
+use anyhow::{Context, Result, bail};
+use futures::stream::{FuturesUnordered, StreamExt};
+use tokio::sync::{Semaphore, broadcast};
+use tracing::{debug, info, instrument, warn};
+
+use crate::drv::DrvGraph;
+
+#[derive(Clone, Debug)]
+pub enum BuildEvent {
+	SubstitutePrepassStarted {
+		paths: usize,
+	},
+	SubstitutePrepassFinished {
+		satisfied: usize,
+	},
+	DrvStarted {
+		drv_path: String,
+		name: String,
+		wanted: Vec<String>,
+	},
+	DrvSkipped {
+		drv_path: String,
+		name: String,
+	},
+	DrvFinished {
+		drv_path: String,
+		name: String,
+	},
+	DrvFailed {
+		drv_path: String,
+		name: String,
+		error: String,
+	},
+	DrvCancelled {
+		drv_path: String,
+		name: String,
+		failed_dep: String,
+	},
+}
+
+pub struct Scheduler {
+	parallelism: usize,
+	events: broadcast::Sender<BuildEvent>,
+}
+
+impl Scheduler {
+	pub fn new(parallelism: usize) -> Self {
+		let parallelism = parallelism.max(1);
+		let (events, _) = broadcast::channel(1024);
+		Self {
+			parallelism,
+			events,
+		}
+	}
+
+	pub fn subscribe(&self) -> broadcast::Receiver<BuildEvent> {
+		self.events.subscribe()
+	}
+
+	#[instrument(name = "scheduler", skip(self, graph), fields(root = %graph.root, nodes = graph.nodes.len()))]
+	pub async fn run(&self, graph: Arc<DrvGraph>, root_outputs: Vec<String>) -> Result<()> {
+		let wanted = graph.wanted_outputs(&root_outputs);
+
+		self.substitute_prepass(&graph, &wanted).await?;
+		self.build_topo(&graph, wanted).await
+	}
+
+	async fn substitute_prepass(
+		&self,
+		graph: &DrvGraph,
+		wanted: &HashMap<String, Vec<String>>,
+	) -> Result<()> {
+		let paths = collect_substitute_paths(graph, wanted);
+		if paths.is_empty() {
+			return Ok(());
+		}
+		let _ = self
+			.events
+			.send(BuildEvent::SubstitutePrepassStarted { paths: paths.len() });
+		debug!("substitute pre-pass: {} paths", paths.len());
+
+		let satisfied = tokio::task::spawn_blocking(move || crate::substitute_paths(&paths))
+			.await
+			.expect("substitute pre-pass task should not panic")?;
+
+		let _ = self.events.send(BuildEvent::SubstitutePrepassFinished {
+			satisfied: satisfied.len(),
+		});
+		Ok(())
+	}
+
+	async fn build_topo(
+		&self,
+		graph: &Arc<DrvGraph>,
+		wanted: HashMap<String, Vec<String>>,
+	) -> Result<()> {
+		let mut indeg: HashMap<String, usize> = graph
+			.nodes
+			.iter()
+			.map(|(k, n)| (k.clone(), n.input_drvs.len()))
+			.collect();
+		let mut dependents: HashMap<String, Vec<String>> = HashMap::new();
+		for (path, node) in &graph.nodes {
+			for dep in node.input_drvs.keys() {
+				dependents
+					.entry(dep.clone())
+					.or_default()
+					.push(path.clone());
+			}
+		}
+
+		let sem = Arc::new(Semaphore::new(self.parallelism));
+		let mut ready: Vec<String> = indeg
+			.iter()
+			.filter(|(_, d)| **d == 0)
+			.map(|(k, _)| k.clone())
+			.collect();
+		let mut in_flight = FuturesUnordered::new();
+		let mut failed: HashMap<String, String> = HashMap::new();
+		// Tainted = transitively depends on a failed drv
+		let mut tainted: HashMap<String, String> = HashMap::new();
+
+		loop {
+			let batch: Vec<String> = std::mem::take(&mut ready);
+			for path in batch {
+				if let Some(failed_dep) = tainted.get(&path) {
+					let name = graph
+						.nodes
+						.get(&path)
+						.map(|n| n.name.clone())
+						.unwrap_or_default();
+					let _ = self.events.send(BuildEvent::DrvCancelled {
+						drv_path: path.clone(),
+						name,
+						failed_dep: failed_dep.clone(),
+					});
+					propagate_done(&dependents, &mut indeg, &mut ready, &path);
+					continue;
+				}
+
+				let sem = sem.clone();
+				let events = self.events.clone();
+				let graph = graph.clone();
+				let wanted_here = wanted.get(&path).cloned().unwrap_or_default();
+				in_flight.push(tokio::spawn(async move {
+					let _permit = sem.acquire_owned().await.expect("semaphore not closed");
+					let node = graph
+						.nodes
+						.get(&path)
+						.expect("ready node must be in graph")
+						.clone();
+					let name = node.name.clone();
+
+					let all_valid = !wanted_here.is_empty()
+						&& wanted_here.iter().all(|o| {
+							node.outputs
+								.get(o)
+								.map(|p| crate::is_valid_path(p).unwrap_or(false))
+								.unwrap_or(false)
+						});
+					if all_valid {
+						let _ = events.send(BuildEvent::DrvSkipped {
+							drv_path: path.clone(),
+							name: name.clone(),
+						});
+						return (path, name, Ok::<(), anyhow::Error>(()));
+					}
+
+					let _ = events.send(BuildEvent::DrvStarted {
+						drv_path: path.clone(),
+						name: name.clone(),
+						wanted: wanted_here.clone(),
+					});
+
+					let path_for_build = path.clone();
+					let res = tokio::task::spawn_blocking(move || {
+						crate::build_drv_outputs(&path_for_build, &wanted_here)
+					})
+					.await
+					.expect("build task should not panic");
+
+					match res {
+						Ok(_) => {
+							let _ = events.send(BuildEvent::DrvFinished {
+								drv_path: path.clone(),
+								name: name.clone(),
+							});
+							(path, name, Ok(()))
+						}
+						Err(e) => {
+							let msg = format!("{e:#}");
+							let _ = events.send(BuildEvent::DrvFailed {
+								drv_path: path.clone(),
+								name: name.clone(),
+								error: msg,
+							});
+							(path, name, Err(e))
+						}
+					}
+				}));
+			}
+
+			let Some(joined) = in_flight.next().await else {
+				break;
+			};
+			let (finished, _name, res) = match joined {
+				Ok(t) => t,
+				Err(e) => bail!("scheduler task panicked: {e}"),
+			};
+			match res {
+				Ok(()) => {
+					propagate_done(&dependents, &mut indeg, &mut ready, &finished);
+				}
+				Err(e) => {
+					failed.insert(finished.clone(), format!("{e:#}"));
+					mark_tainted(&dependents, &finished, &mut tainted);
+					propagate_done(&dependents, &mut indeg, &mut ready, &finished);
+				}
+			}
+		}
+
+		let stuck: Vec<_> = indeg
+			.iter()
+			.filter(|(_, d)| **d != 0)
+			.map(|(k, _)| k.as_str())
+			.collect();
+		if !stuck.is_empty() {
+			warn!(
+				"scheduler finished with {} nodes still pending (loop?)",
+				stuck.len()
+			);
+		}
+
+		if failed.is_empty() {
+			info!("scheduler completed");
+			Ok(())
+		} else {
+			let mut report = format!("{} drv(s) failed to build:", failed.len());
+			let mut sorted: Vec<_> = failed.iter().collect();
+			sorted.sort_by(|a, b| a.0.cmp(b.0));
+			for (path, err) in sorted {
+				let name = graph
+					.nodes
+					.get(path)
+					.map(|n| n.name.as_str())
+					.unwrap_or("?");
+				let chain = path_to_root(graph, path);
+				report.push_str(&format!(
+					"\n\n  {name} ({path}):\n    {err}\n    needed by: {}",
+					chain.join(" => "),
+				));
+			}
+			Err(anyhow::anyhow!(report))
+		}
+	}
+}
+
+fn propagate_done(
+	dependents: &HashMap<String, Vec<String>>,
+	indeg: &mut HashMap<String, usize>,
+	ready: &mut Vec<String>,
+	finished: &str,
+) {
+	if let Some(deps) = dependents.get(finished) {
+		for d in deps {
+			let entry = indeg.get_mut(d).expect("dependent must have indeg");
+			*entry = entry.saturating_sub(1);
+			if *entry == 0 {
+				ready.push(d.clone());
+			}
+		}
+	}
+}
+
+fn mark_tainted(
+	dependents: &HashMap<String, Vec<String>>,
+	failed: &str,
+	tainted: &mut HashMap<String, String>,
+) {
+	let mut queue: Vec<String> = dependents.get(failed).cloned().unwrap_or_default();
+	while let Some(node) = queue.pop() {
+		if tainted
+			.entry(node.clone())
+			.or_insert_with(|| failed.to_owned())
+			== failed
+		{
+			if let Some(deps) = dependents.get(&node) {
+				for d in deps {
+					if !tainted.contains_key(d) {
+						queue.push(d.clone());
+					}
+				}
+			}
+		}
+	}
+}
+
+fn path_to_root(graph: &DrvGraph, from: &str) -> Vec<String> {
+	let mut dependents: HashMap<&str, Vec<&str>> = HashMap::new();
+	for (path, node) in &graph.nodes {
+		for dep in node.input_drvs.keys() {
+			dependents
+				.entry(dep.as_str())
+				.or_default()
+				.push(path.as_str());
+		}
+	}
+
+	let mut chain: Vec<String> = vec![node_name(graph, from)];
+	let mut cur = from;
+	let mut seen: HashSet<&str> = HashSet::new();
+	seen.insert(cur);
+	while cur != graph.root.as_str() {
+		let Some(next) = dependents.get(cur).and_then(|v| v.first().copied()) else {
+			break;
+		};
+		if !seen.insert(next) {
+			break;
+		}
+		chain.push(node_name(graph, next));
+		cur = next;
+	}
+	chain
+}
+
+fn node_name(graph: &DrvGraph, path: &str) -> String {
+	graph
+		.nodes
+		.get(path)
+		.map(|n| n.name.clone())
+		.unwrap_or_else(|| path.to_owned())
+}
+
+fn collect_substitute_paths(
+	graph: &DrvGraph,
+	wanted: &HashMap<String, Vec<String>>,
+) -> Vec<String> {
+	let mut paths: HashSet<String> = HashSet::new();
+	for node in graph.nodes.values() {
+		for src in &node.input_srcs {
+			paths.insert(src.clone());
+		}
+	}
+	for (path, outs) in wanted {
+		let Some(node) = graph.nodes.get(path) else {
+			continue;
+		};
+		for o in outs {
+			if let Some(p) = node.outputs.get(o) {
+				paths.insert(p.clone());
+			}
+		}
+	}
+	let mut v: Vec<_> = paths.into_iter().collect();
+	v.sort();
+	v
+}
+
+// TODO: Parallelism as a metric works poorly with multiple machines, but I haven't thought about bringing
+// hercy here yet. In case of remote machines - they will handle parallelism on their own, and this one
+// will work as a hard cap.
+pub fn build_graph_sync(graph: Arc<DrvGraph>, root_outputs: Vec<String>) -> Result<()> {
+	let parallelism = std::thread::available_parallelism()
+		.map(|p| p.get())
+		.unwrap_or(4);
+	let scheduler = Scheduler::new(parallelism);
+	crate::await_in_nix(async move { scheduler.run(graph, root_outputs).await })
+		.context("scheduler run")
+}
modifiedcrates/nixlike/Cargo.tomldiffbeforeafterboth
--- a/crates/nixlike/Cargo.toml
+++ b/crates/nixlike/Cargo.toml
@@ -11,5 +11,3 @@
 linked-hash-map.workspace = true
 peg.workspace = true
 serde.workspace = true
-serde-transcode.workspace = true
-serde_json.workspace = true
addedcrates/remowt-fleet/Cargo.tomldiffbeforeafterboth
--- /dev/null
+++ b/crates/remowt-fleet/Cargo.toml
@@ -0,0 +1,20 @@
+[package]
+name = "remowt-fleet"
+description = "Remowt endpoints for fleet"
+version.workspace = true
+edition = "2021"
+
+[dependencies]
+bifrostlink.workspace = true
+
+remowt-client.workspace = true
+remowt-endpoints.workspace = true
+
+anyhow.workspace = true
+camino.workspace = true
+nix-eval.workspace = true
+serde = { workspace = true }
+thiserror.workspace = true
+tokio = { workspace = true, features = ["io-util", "net", "process", "rt"] }
+tracing.workspace = true
+uuid.workspace = true
addedcrates/remowt-fleet/src/lib.rsdiffbeforeafterboth
--- /dev/null
+++ b/crates/remowt-fleet/src/lib.rs
@@ -0,0 +1,120 @@
+use std::path::PathBuf;
+
+use anyhow::{Context as _, Result};
+use bifrostlink::Config;
+use bifrostlink::declarative::endpoints;
+use camino::Utf8PathBuf;
+use remowt_client::Remowt;
+use remowt_endpoints::nix_daemon::NixDaemonClient;
+use serde::{Deserialize, Serialize};
+use tokio::net::UnixListener;
+use tracing::error;
+
+pub struct Nix;
+pub use nix_eval::{init_libraries, init_tokio_for_nix};
+
+#[derive(Serialize, Deserialize, Debug, thiserror::Error)]
+pub enum NixError {
+	#[error("nix daemon unavailable: {0}")]
+	DaemonUnavailable(String),
+	#[error("tunnel socket unavailable: {0}")]
+	Tunnel(String),
+	#[error("profile switch failed: {0}")]
+	Profile(String),
+	#[error("signing failed: {0}")]
+	Sign(String),
+	#[error("listing generations failed: {0}")]
+	ListGenerations(String),
+}
+
+#[endpoints(ns = 91)]
+impl Nix {
+	#[endpoints(id = 3)]
+	async fn switch_profile(
+		&self,
+		profile: String,
+		store_path: Utf8PathBuf,
+	) -> Result<(), NixError> {
+		tokio::task::spawn_blocking(move || nix_eval::switch_profile(&profile, &store_path))
+			.await
+			.map_err(|e| NixError::Profile(e.to_string()))?
+			.map_err(|e| NixError::Profile(e.to_string()))
+	}
+
+	#[endpoints(id = 4)]
+	async fn sign_closure(
+		&self,
+		store_path: Utf8PathBuf,
+		key_file: Utf8PathBuf,
+	) -> Result<(), NixError> {
+		tokio::task::spawn_blocking(move || {
+			nix_eval::sign_closure(store_path.as_str(), key_file.as_str())
+		})
+		.await
+		.map_err(|e| NixError::Sign(e.to_string()))?
+		.map_err(|e| NixError::Sign(e.to_string()))
+	}
+
+	#[endpoints(id = 5)]
+	async fn list_generations(
+		&self,
+		profile: String,
+	) -> Result<Vec<nix_eval::ProfileGeneration>, NixError> {
+		tokio::task::spawn_blocking(move || {
+			nix_eval::list_generations(&format!("/nix/var/nix/profiles/{profile}"))
+		})
+		.await
+		.map_err(|e| NixError::ListGenerations(e.to_string()))?
+		.map_err(|e| NixError::ListGenerations(e.to_string()))
+	}
+}
+
+pub async fn nix_store_socket(conn: Remowt, store: &str) -> Result<PathBuf> {
+	let store = store.to_owned();
+	let path = std::env::temp_dir().join(format!("fleet-nix-{}.sock", uuid::Uuid::new_v4()));
+	let _ = std::fs::remove_file(&path);
+	let listener = UnixListener::bind(&path)?;
+	tokio::spawn(async move {
+		if let Err(e) = serve(conn, listener, store).await {
+			error!("nix daemon proxy failed: {e}");
+		}
+	});
+	Ok(path)
+}
+
+async fn serve(conn: Remowt, listener: UnixListener, store: String) -> Result<()> {
+	let nix = conn.endpoints::<NixDaemonClient<_>>();
+	loop {
+		let (mut local, _) = listener.accept().await?;
+
+		let (rx, remote_sock) = match conn.bind_runtime_unix("nix-daemon").await {
+			Ok(rx) => rx,
+			Err(e) => {
+				error!("streamlocal_forward failed: {e}");
+				continue;
+			}
+		};
+		let sock_str = remote_sock.as_str().to_owned();
+		match nix.serve_store(store.clone(), sock_str).await {
+			Ok(Ok(())) => {}
+			Ok(Err(e)) => {
+				error!("nix bridge: {e}");
+				continue;
+			}
+			Err(e) => {
+				error!("nix bridge rpc failed: {e}");
+				continue;
+			}
+		}
+
+		let mut channel = rx
+			.accept()
+			.await
+			.context("failed to accept remote nix connection")?;
+		tokio::spawn(async move {
+			if let Err(e) = tokio::io::copy_bidirectional(&mut local, &mut channel).await {
+				tracing::debug!("nix tunnel ended: {e}");
+			}
+		});
+	}
+}
modifiedflake.lockdiffbeforeafterboth
--- a/flake.lock
+++ b/flake.lock
@@ -2,10 +2,10 @@
   "nodes": {
     "crane": {
       "locked": {
-        "lastModified": 1773189535,
+        "lastModified": 1780532242,
         "owner": "ipetkov",
         "repo": "crane",
-        "rev": "6fa2fb4cf4a89ba49fc9dd5a3eb6cde99d388269",
+        "rev": "59a82a1222dd3b2080b5cc52a1a2e8d5f1b77f37",
         "type": "github"
       },
       "original": {
@@ -37,10 +37,10 @@
         ]
       },
       "locked": {
-        "lastModified": 1772408722,
+        "lastModified": 1778716662,
         "owner": "hercules-ci",
         "repo": "flake-parts",
-        "rev": "f20dc5d9b8027381c474144ecabc9034d6a839a3",
+        "rev": "f7c1a2d347e4c52d5fb8d10cb4d94b5884e546fb",
         "type": "github"
       },
       "original": {
@@ -62,10 +62,10 @@
         ]
       },
       "locked": {
-        "lastModified": 1759080490,
+        "lastModified": 1780160585,
         "owner": "CertainLach",
         "repo": "fleet-tf",
-        "rev": "878bd8c23933d628bf750378bbe527b841901c3d",
+        "rev": "aa2200a3f8f93a8c0d3b87bb29f11d9d9b0a23e6",
         "type": "github"
       },
       "original": {
@@ -111,10 +111,10 @@
         "nixpkgs-regression": "nixpkgs-regression"
       },
       "locked": {
-        "lastModified": 1768702010,
+        "lastModified": 1781448983,
         "owner": "deltarocks",
         "repo": "nix",
-        "rev": "b05b52670b9c7affff5b9be3edb539a1603c39e6",
+        "rev": "bb8784f285658e77642437409824619f5c34a854",
         "type": "github"
       },
       "original": {
@@ -126,10 +126,10 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1773268153,
+        "lastModified": 1781402740,
         "owner": "nixos",
         "repo": "nixpkgs",
-        "rev": "63ca4f465b79edbaa9ff113ea7ff48fb49ea0db0",
+        "rev": "49145c5aba740948d6da1265f584826bd27619ad",
         "type": "github"
       },
       "original": {
@@ -171,6 +171,34 @@
         "type": "github"
       }
     },
+    "remowt-agents": {
+      "inputs": {
+        "crane": [
+          "crane"
+        ],
+        "flake-parts": [
+          "flake-parts"
+        ],
+        "nixpkgs": [
+          "nixpkgs"
+        ],
+        "rust-overlay": "rust-overlay",
+        "shelly": [
+          "shelly"
+        ]
+      },
+      "locked": {
+        "lastModified": 1781471358,
+        "rev": "c69dddc33f8e2081ef685ae20a7278f92cfcd32e",
+        "revCount": 37,
+        "type": "git",
+        "url": "file:/home/lach/build/remowt-agents"
+      },
+      "original": {
+        "type": "git",
+        "url": "file:/home/lach/build/remowt-agents"
+      }
+    },
     "root": {
       "inputs": {
         "crane": "crane",
@@ -178,7 +206,8 @@
         "fleet-tf": "fleet-tf",
         "nix": "nix",
         "nixpkgs": "nixpkgs",
-        "rust-overlay": "rust-overlay",
+        "remowt-agents": "remowt-agents",
+        "rust-overlay": "rust-overlay_2",
         "shelly": "shelly",
         "treefmt-nix": "treefmt-nix"
       }
@@ -186,14 +215,15 @@
     "rust-overlay": {
       "inputs": {
         "nixpkgs": [
+          "remowt-agents",
           "nixpkgs"
         ]
       },
       "locked": {
-        "lastModified": 1773216618,
+        "lastModified": 1769309768,
         "owner": "oxalica",
         "repo": "rust-overlay",
-        "rev": "07d7dc6fcc5eae76b4fb0e19d4afd939437bec97",
+        "rev": "140c9dc582cb73ada2d63a2180524fcaa744fad5",
         "type": "github"
       },
       "original": {
@@ -202,6 +232,25 @@
         "type": "github"
       }
     },
+    "rust-overlay_2": {
+      "inputs": {
+        "nixpkgs": [
+          "nixpkgs"
+        ]
+      },
+      "locked": {
+        "lastModified": 1781320681,
+        "owner": "oxalica",
+        "repo": "rust-overlay",
+        "rev": "5b929d8c854149d926d05ea0cd6469bf4e54db27",
+        "type": "github"
+      },
+      "original": {
+        "owner": "oxalica",
+        "repo": "rust-overlay",
+        "type": "github"
+      }
+    },
     "shelly": {
       "locked": {
         "lastModified": 1756323923,
@@ -223,10 +272,10 @@
         ]
       },
       "locked": {
-        "lastModified": 1772660329,
+        "lastModified": 1780220602,
         "owner": "numtide",
         "repo": "treefmt-nix",
-        "rev": "3710e0e1218041bbad640352a0440114b1e10428",
+        "rev": "db947814a175b7ca6ded66e21383d938df01c227",
         "type": "github"
       },
       "original": {
modifiedflake.nixdiffbeforeafterboth
--- a/flake.nix
+++ b/flake.nix
@@ -23,6 +23,13 @@
       url = "github:numtide/treefmt-nix";
       inputs.nixpkgs.follows = "nixpkgs";
     };
+    remowt-agents = {
+      url = "git+file:/home/lach/build/remowt-agents";
+      inputs.nixpkgs.follows = "nixpkgs";
+      inputs.flake-parts.follows = "flake-parts";
+      inputs.crane.follows = "crane";
+      inputs.shelly.follows = "shelly";
+    };
     # DeterminateSystem's nix fork is controversial, but I don't mind it,
     # and it has lazy-trees support which is useful for fleet.
     nix = {
@@ -147,7 +154,7 @@
             packages = lib.mkIf deployerSystem (
               let
                 packages = pkgs.callPackages ./pkgs {
-                  inherit craneLib inputs';
+                  inherit craneLib inputs;
                 };
               in
               packages // { default = packages.fleet; }
modifiedlib/flakePart.nixdiffbeforeafterboth
--- a/lib/flakePart.nix
+++ b/lib/flakePart.nix
@@ -57,7 +57,7 @@
             modules = (import ../modules/module-list.nix) ++ [
               module
               (
-                { inputs', ... }:
+                { inputs, ... }:
                 {
                   config = {
                     data = if isPath data then import data else data;
@@ -66,12 +66,13 @@
                       (final: prev: {
                         inherit
                           (import ../pkgs {
-                            inherit (prev) callPackage;
-                            inherit inputs';
+                            inherit (final) callPackage;
+                            inherit inputs;
                             craneLib = crane.mkLib prev;
                           })
                           fleet-install-secrets
                           fleet-generator-helper
+                          remowt-plugin-fleet
                           ;
                       })
                     ];
modifiedmodules/nixpkgs.nixdiffbeforeafterboth
--- a/modules/nixpkgs.nix
+++ b/modules/nixpkgs.nix
@@ -42,10 +42,12 @@
         default = config.nixpkgs.buildUsing;
         defaultText = literalExpression "config.nixpkgs.buildUsing";
       };
-      config.nixos = {
-        inherit _file;
-        nixpkgs.overlays = config.nixpkgs.overlays;
-      };
+      config.nixos =
+        { inputs', ... }:
+        {
+          inherit _file;
+          nixpkgs.overlays = config.nixpkgs.overlays;
+        };
     };
   };
 }
modifiedpkgs/default.nixdiffbeforeafterboth
--- a/pkgs/default.nix
+++ b/pkgs/default.nix
@@ -1,10 +1,11 @@
 {
   callPackage,
   craneLib,
-  inputs',
+  inputs,
 }:
 {
-  fleet = callPackage ./fleet.nix { inherit craneLib inputs'; };
+  fleet = callPackage ./fleet.nix { inherit craneLib inputs; };
+  remowt-plugin-fleet = callPackage ./remowt-plugin-fleet.nix { inherit craneLib inputs; };
   fleet-install-secrets = callPackage ./fleet-install-secrets.nix { inherit craneLib; };
   fleet-generator-helper = callPackage ./fleet-generator-helper.nix { inherit craneLib; };
 }
modifiedpkgs/fleet.nixdiffbeforeafterboth
--- a/pkgs/fleet.nix
+++ b/pkgs/fleet.nix
@@ -2,10 +2,16 @@
   lib,
   craneLib,
   installShellFiles,
-  inputs',
+  inputs,
+
+  stdenv,
   pkg-config,
   rustPlatform,
+  rofi,
 }:
+let
+  system = stdenv.hostPlatform.system;
+in
 craneLib.buildPackage rec {
   pname = "fleet";
   src = lib.cleanSourceWith {
@@ -20,10 +26,15 @@
 
   cargoExtraArgs = "--locked -p ${pname}";
 
+  REMOWT_AGENTS_DIR = "${inputs.remowt-agents.packages.${system}.remowt-agents}";
+  # TODO: built-in fleet prompter should be a prodash widget, or it should require
+  # tty remowt prompter running on host machine idk.
+  ROFI = "${rofi}/bin/rofi";
+
   buildInputs = [
-    inputs'.nix.packages.nix-expr-c
-    inputs'.nix.packages.nix-flake-c
-    inputs'.nix.packages.nix-fetchers-c
+    inputs.nix.packages.${system}.nix-expr-c
+    inputs.nix.packages.${system}.nix-flake-c
+    inputs.nix.packages.${system}.nix-fetchers-c
   ];
   nativeBuildInputs = [
     installShellFiles
addedpkgs/remowt-plugin-fleet.nixdiffbeforeafterboth
--- /dev/null
+++ b/pkgs/remowt-plugin-fleet.nix
@@ -0,0 +1,36 @@
+{
+  lib,
+  craneLib,
+  inputs,
+
+  stdenv,
+  pkg-config,
+  rustPlatform,
+}:
+let
+  system = stdenv.hostPlatform.system;
+in
+craneLib.buildPackage rec {
+  pname = "remowt-plugin-fleet";
+  src = lib.cleanSourceWith {
+    src = ../.;
+    filter =
+      path: type:
+      (lib.hasSuffix "\.cc" path)
+      || (lib.hasSuffix "\.hh" path)
+      || (craneLib.filterCargoSources path type);
+  };
+  strictDeps = true;
+
+  cargoExtraArgs = "--locked -p ${pname}";
+
+  buildInputs = [
+    inputs.nix.packages.${system}.nix-expr-c
+    inputs.nix.packages.${system}.nix-flake-c
+    inputs.nix.packages.${system}.nix-fetchers-c
+  ];
+  nativeBuildInputs = [
+    pkg-config
+    rustPlatform.bindgenHook
+  ];
+}