Skip to main content

qk_executor/
lib.rs

1//! Prepare finite tasks, then execute shell commands in managed process groups.
2
3mod capture;
4pub mod dotenv;
5mod env_flags;
6mod interpolate;
7mod process;
8#[cfg(unix)]
9mod pty;
10pub mod readiness;
11pub mod ready_signal;
12pub mod report;
13pub mod tokens;
14
15use std::collections::BTreeMap;
16use std::ffi::OsString;
17use std::path::{Path, PathBuf};
18
19use anyhow::{Context, Result, bail};
20use qk_config::Workspace;
21use qk_taskgraph::Task;
22use serde_json::{Map, Value};
23
24pub use capture::{
25    Capture, Decoration, Display, MAX_LOG_BYTES, OutputStyle, RecordedLog, Shown, collapsed,
26    printable, read_capture, replay,
27};
28pub use env_flags::{force_colour, is_ci, is_ci_with};
29pub use process::{Outcome, execute, execute_captured, shell};
30
31#[derive(Debug)]
32struct NodeRuntime {
33    root: PathBuf,
34    env: BTreeMap<OsString, OsString>,
35    paths: Vec<PathBuf>,
36    directory: std::sync::OnceLock<Option<PathBuf>>,
37}
38
39/// The runner's Node selection, resolved only when a command starts.
40#[derive(Clone, Debug)]
41pub struct NodePath {
42    runtime: std::sync::Arc<NodeRuntime>,
43    local_entries: usize,
44}
45
46impl NodePath {
47    fn new(root: &Path, env: &BTreeMap<OsString, OsString>, local_entries: usize) -> Self {
48        type Key = (PathBuf, BTreeMap<OsString, OsString>);
49        static RUNTIMES: std::sync::OnceLock<
50            std::sync::Mutex<BTreeMap<Key, std::sync::Arc<NodeRuntime>>>,
51        > = std::sync::OnceLock::new();
52        let mut runtimes = RUNTIMES.get_or_init(Default::default).lock().unwrap();
53        let runtime = runtimes
54            .entry((root.to_owned(), env.clone()))
55            .or_insert_with(|| {
56                let paths = env
57                    .get(&path_key(env))
58                    .map(|path| std::env::split_paths(path).collect())
59                    .unwrap_or_default();
60                std::sync::Arc::new(NodeRuntime {
61                    root: root.to_owned(),
62                    env: env.clone(),
63                    paths,
64                    directory: Default::default(),
65                })
66            })
67            .clone();
68        Self {
69            runtime,
70            local_entries,
71        }
72    }
73
74    fn path(&self, env: &BTreeMap<OsString, OsString>) -> Result<Option<(OsString, OsString)>> {
75        // Select before task overrides, like process.execPath in Nx. Cache hits
76        // never call this, and concurrent commands share the same proxy probe.
77        let Some(directory) = self.runtime.directory.get_or_init(|| {
78            node_directory(&self.runtime.root, &self.runtime.paths, &self.runtime.env)
79        }) else {
80            return Ok(None);
81        };
82        let key = path_key(env);
83        let mut paths: Vec<_> = env
84            .get(&key)
85            .map(|path| std::env::split_paths(path).collect())
86            .unwrap_or_default();
87        paths.insert(self.local_entries.min(paths.len()), directory.clone());
88        Ok(Some((
89            key,
90            std::env::join_paths(paths).context("cannot construct task PATH")?,
91        )))
92    }
93}
94
95fn path_key(env: &BTreeMap<OsString, OsString>) -> OsString {
96    if cfg!(windows) {
97        env.keys()
98            .find(|key| key.to_string_lossy().eq_ignore_ascii_case("PATH"))
99            .cloned()
100            .unwrap_or_else(|| "PATH".into())
101    } else {
102        "PATH".into()
103    }
104}
105
106fn environment_key(env: &BTreeMap<OsString, OsString>, name: OsString) -> OsString {
107    if cfg!(windows) && name.to_string_lossy().eq_ignore_ascii_case("PATH") {
108        path_key(env)
109    } else {
110        name
111    }
112}
113
114fn node_directory(
115    cwd: &Path,
116    paths: &[PathBuf],
117    env: &BTreeMap<OsString, OsString>,
118) -> Option<PathBuf> {
119    let name = if cfg!(windows) { "node.exe" } else { "node" };
120    for directory in paths {
121        let candidate = cwd.join(directory).join(name);
122        let Ok(metadata) = candidate.metadata() else {
123            continue;
124        };
125        if !metadata.is_file() {
126            continue;
127        }
128        #[cfg(unix)]
129        {
130            use std::os::unix::fs::PermissionsExt;
131            if metadata.permissions().mode() & 0o111 == 0 {
132                continue;
133            }
134        }
135        // The first executable PATH match is authoritative, even if discovery fails.
136        let actual = candidate.canonicalize().ok()?;
137        if actual.file_name()? == name {
138            use std::io::Read;
139            let mut header = [0; 2];
140            let script = std::fs::File::open(&actual)
141                .and_then(|mut file| file.read_exact(&mut header))
142                .is_ok()
143                && header == *b"#!";
144            if !script {
145                return actual.parent().map(Path::to_owned);
146            }
147        }
148        return probe_node(&candidate, cwd, env);
149    }
150    None
151}
152
153fn probe_node(candidate: &Path, cwd: &Path, env: &BTreeMap<OsString, OsString>) -> Option<PathBuf> {
154    use std::io::{Read, Seek};
155    use std::process::{Command, Stdio};
156    use std::time::{Duration, Instant};
157    let mut output = tempfile::tempfile().ok()?;
158    let mut child = Command::new(candidate)
159        .args(["-p", "process.execPath"])
160        .current_dir(cwd)
161        .env_clear()
162        .envs(env)
163        .stdout(Stdio::from(output.try_clone().ok()?))
164        .stderr(Stdio::null())
165        .spawn()
166        .ok()?;
167    let started = Instant::now();
168    loop {
169        match child.try_wait() {
170            Ok(Some(status)) => {
171                if !status.success() {
172                    return None;
173                }
174                let mut bytes = Vec::new();
175                output.rewind().ok()?;
176                output.take(4096).read_to_end(&mut bytes).ok()?;
177                let path = PathBuf::from(std::str::from_utf8(&bytes).ok()?.trim());
178                return (path.is_absolute() && path.is_file())
179                    .then(|| path.parent().unwrap().to_owned());
180            }
181            Ok(None) if started.elapsed() < Duration::from_secs(2) => {
182                std::thread::sleep(Duration::from_millis(5))
183            }
184            _ => {
185                let _ = child.kill();
186                let _ = child.wait();
187                return None;
188            }
189        }
190    }
191}
192
193/// Reports a `qk:` warning to the installed sink; by default it is written to
194/// stderr in one write, so task processes sharing stderr cannot split it.
195#[macro_export]
196macro_rules! status {
197    ($($arg:tt)*) => {{
198        $crate::report::warning(&::std::format!($($arg)*));
199    }};
200}
201
202/// How a sandboxed task's commands are held to what it declares.
203#[derive(Clone, Debug)]
204pub enum Confinement {
205    /// macOS: a Seatbelt profile, through `sandbox-exec`.
206    Seatbelt(PathBuf),
207    /// Linux: a Landlock rule set's descriptor, which each command restricts
208    /// itself to before it starts. It stays open while the run lasts.
209    Landlock(i32),
210    /// Enforced, and could not be held to its plan, for this reason: its
211    /// commands fail rather than start unconfined.
212    Refused(String),
213}
214
215/// An external syscall recorder wrapping each command, with separate trace files.
216#[derive(Clone, Debug)]
217pub struct Recording {
218    /// The recorder's program: strace, or qk itself, whose helper loads the
219    /// eBPF recorder.
220    pub program: PathBuf,
221    pub directory: PathBuf,
222    /// Whether strace filters the traced calls in the kernel, with seccomp,
223    /// so untraced calls run at full speed. strace 5.3 and later can.
224    pub seccomp: bool,
225    /// The eBPF recorder, and the recordings of this task's commands it has
226    /// begun; `None` when strace records.
227    #[cfg(target_os = "linux")]
228    ebpf: Option<Ebpf>,
229}
230
231#[cfg(target_os = "linux")]
232#[derive(Clone, Debug)]
233struct Ebpf {
234    recorder: std::sync::Arc<qk_trace::Recorder>,
235    sessions: std::sync::Arc<std::sync::Mutex<Vec<qk_trace::Session>>>,
236    /// Why settling failed, kept so that a later settle, as the cache's and
237    /// then an audit's of one recording, does not find it whole.
238    failure: std::sync::Arc<std::sync::Mutex<Option<String>>>,
239    /// Whether an opened file's path is recorded with what its links lead
240    /// to, as strace records it, which costs a lookup an open.
241    resolve: bool,
242}
243
244/// The eBPF recorder, its helper started once a process: as qk itself when
245/// qk runs as root, else through `sudo -n`, so that it is never asked for a
246/// password.
247#[cfg(target_os = "linux")]
248fn ebpf_recorder() -> Result<std::sync::Arc<qk_trace::Recorder>, String> {
249    static RECORDER: std::sync::OnceLock<Result<std::sync::Arc<qk_trace::Recorder>, String>> =
250        std::sync::OnceLock::new();
251    RECORDER
252        .get_or_init(|| {
253            let qk = std::env::current_exe().map_err(|error| error.to_string())?;
254            // SAFETY: geteuid has no preconditions.
255            let helper = if unsafe { libc::geteuid() } == 0 {
256                let mut helper = std::process::Command::new(qk);
257                helper.arg("__trace-helper");
258                helper
259            } else {
260                let mut helper = std::process::Command::new("sudo");
261                helper.args(["-n", "--"]).arg(qk).arg("__trace-helper");
262                helper
263            };
264            qk_trace::Recorder::start(helper).map_err(|error| format!("{error:#}"))
265        })
266        .clone()
267}
268
269impl Recording {
270    /// The recorder `QK_RECORDER` picks: by default eBPF, where its helper
271    /// can load it, else strace on `PATH`, checked to record child
272    /// processes, with seccomp filtering where it supports it. Its trial
273    /// traces go in `directory`.
274    pub fn probe(directory: &Path) -> anyhow::Result<Self> {
275        let choice = std::env::var("QK_RECORDER").unwrap_or_default();
276        match choice.as_str() {
277            "" | "auto" | "ebpf" =>
278            {
279                #[cfg(target_os = "linux")]
280                match ebpf_recorder() {
281                    Ok(recorder) => {
282                        return Ok(Self {
283                            program: std::env::current_exe()?,
284                            directory: directory.to_owned(),
285                            seccomp: false,
286                            ebpf: Some(Ebpf {
287                                recorder,
288                                sessions: Default::default(),
289                                failure: Default::default(),
290                                resolve: false,
291                            }),
292                        });
293                    }
294                    Err(error) if choice == "ebpf" => {
295                        anyhow::bail!("cannot record with eBPF: {error}")
296                    }
297                    Err(_) => {}
298                }
299            }
300            "strace" => {}
301            other => anyhow::bail!("unknown QK_RECORDER {other:?}; use auto, ebpf or strace"),
302        }
303        let path = std::env::var_os("PATH").context("recording needs strace on PATH")?;
304        let program = std::env::split_paths(&path)
305            .map(|directory| directory.join("strace"))
306            .find(|candidate| candidate.is_file())
307            .context("recording needs strace on PATH; install strace and retry")?
308            .canonicalize()?;
309        let mut failure = String::new();
310        for seccomp in [true, false] {
311            let recording = Self {
312                program: program.clone(),
313                directory: directory.to_owned(),
314                seccomp,
315                #[cfg(target_os = "linux")]
316                ebpf: None,
317            };
318            let probe = recording
319                .command(0, "exec /bin/true", Path::new("/"))
320                .output()
321                .context("cannot probe strace")?;
322            if probe.status.success() {
323                return Ok(recording);
324            }
325            failure = String::from_utf8_lossy(&probe.stderr).into_owned();
326        }
327        anyhow::bail!("strace cannot record child processes: {failure}")
328    }
329
330    /// The same settings, with traces in `directory`.
331    /// The recorder's name, as reports give it: `ebpf` or `strace`.
332    pub fn backend(&self) -> &'static str {
333        #[cfg(target_os = "linux")]
334        if self.ebpf.is_some() {
335            return "ebpf";
336        }
337        "strace"
338    }
339
340    pub fn in_directory(&self, directory: PathBuf) -> Self {
341        Self {
342            directory,
343            #[cfg(target_os = "linux")]
344            ebpf: self.ebpf.as_ref().map(|ebpf| Ebpf {
345                recorder: ebpf.recorder.clone(),
346                sessions: Default::default(),
347                failure: Default::default(),
348                resolve: ebpf.resolve,
349            }),
350            ..self.clone()
351        }
352    }
353
354    /// The same, recording an opened file's path with what its links lead
355    /// to, as a report of what a task read shows it. strace always does.
356    pub fn resolving(self) -> Self {
357        #[cfg(target_os = "linux")]
358        {
359            let mut recording = self;
360            if let Some(ebpf) = &mut recording.ebpf {
361                ebpf.resolve = true;
362            }
363            recording
364        }
365        #[cfg(not(target_os = "linux"))]
366        self
367    }
368
369    /// Ends the recordings of the commands that have run, once they have
370    /// finished, so that their traces are whole. strace's are as soon as it
371    /// exits; the eBPF recorder's once its helper has sent every event.
372    pub fn settle(&self) -> anyhow::Result<()> {
373        #[cfg(target_os = "linux")]
374        if let Some(ebpf) = &self.ebpf {
375            if let Some(failure) = ebpf.failure.lock().unwrap().as_ref() {
376                anyhow::bail!("{failure}");
377            }
378            let sessions = std::mem::take(&mut *ebpf.sessions.lock().unwrap());
379            let mut sessions = sessions.into_iter();
380            while let Some(session) = sessions.next() {
381                if let Err(error) = ebpf.recorder.finish(session) {
382                    *ebpf.failure.lock().unwrap() = Some(format!("{error:#}"));
383                    // The rest without waiting again: a helper that did not
384                    // answer once will not now.
385                    sessions
386                        .by_ref()
387                        .for_each(|rest| ebpf.recorder.abandon(rest));
388                    return Err(error);
389                }
390            }
391        }
392        Ok(())
393    }
394
395    /// The same invocation is used by preflight and actual task commands,
396    /// which start in `cwd`.
397    pub fn command(&self, index: usize, text: &str, cwd: &Path) -> std::process::Command {
398        #[cfg(target_os = "linux")]
399        if let Some(ebpf) = &self.ebpf {
400            use std::os::unix::process::CommandExt;
401            let session = ebpf.recorder.begin(
402                &self.directory,
403                &format!("command-{index}"),
404                cwd,
405                ebpf.resolve,
406            );
407            ebpf.sessions.lock().unwrap().push(session);
408            let mut command = crate::process::shell(text);
409            // SAFETY: the hook makes one system call.
410            unsafe { command.pre_exec(ebpf.recorder.join(session)) };
411            return command;
412        }
413        let _ = cwd;
414        let mut command = std::process::Command::new(&self.program);
415        // No read/write buffers or exec argument/environment arrays.
416        if self.seccomp {
417            command.arg("--seccomp-bpf");
418        }
419        command
420            .args([
421                "-ff",
422                "-ttt",
423                "-yy",
424                "-qq",
425                "-s",
426                "4096",
427                "-e",
428                "trace=%file,%process,getdents64,fchdir,close,dup,dup2,dup3,getcwd",
429                "-e",
430                "raw=execve,execveat",
431                "-o",
432            ])
433            .arg(self.directory.join(format!("command-{index}")))
434            .args(["--", "/bin/sh", "-c", text]);
435        command
436    }
437}
438
439/// A scratch directory, below the workspace `root` it lies in, which the
440/// task keeps nothing in between runs.
441#[derive(Clone, Debug)]
442pub struct Scratch {
443    pub root: PathBuf,
444    pub directory: PathBuf,
445}
446
447#[derive(Clone, Debug)]
448pub struct PreparedTask {
449    pub id: String,
450    pub commands: Vec<String>,
451    pub parallel: bool,
452    pub cwd: PathBuf,
453    pub env: BTreeMap<OsString, OsString>,
454    /// Dotenv, explicit settings and local binary lookup without task metadata.
455    pub runtime_env: BTreeMap<OsString, OsString>,
456    pub node_path: Option<NodePath>,
457    /// Variables set for the process after `env` that are not part of what
458    /// the task is: nothing reads them to key it, such as a thread count.
459    pub execution: BTreeMap<OsString, OsString>,
460    /// Nx's `readyWhen`: text whose appearance in the output, every piece of
461    /// it, makes the task ready while its commands keep running.
462    pub ready_when: Vec<String>,
463    /// Set once all of `ready_when` has appeared.
464    pub ready: std::sync::Arc<std::sync::atomic::AtomicBool>,
465    /// A continuous target's own `readyWhen`, which tasks depending on it
466    /// with `waitFor: "ready"` wait for, or why it is invalid. Prepared, it is
467    /// resolved without `NX_READY_URL`; see [`PreparedTask::resolve_ready_check`].
468    pub ready_check: Option<std::result::Result<readiness::ReadyCheck, String>>,
469    /// What `ready_check` is resolved from, to resolve it again once the
470    /// servers the task waits for have reported their address.
471    pub ready_inputs: Option<std::sync::Arc<ReadyInputs>>,
472    /// What `ready_check` concluded.
473    pub verdict: std::sync::Arc<readiness::Verdict>,
474    /// The process ids of its commands while they run, from which the runner
475    /// measures the memory the task uses.
476    pub processes: std::sync::Arc<std::sync::Mutex<std::collections::BTreeSet<u32>>>,
477    /// Each command's prefix and colours, by position; empty without any.
478    pub decorations: Vec<Decoration>,
479    /// Whether the task reads qk's stdin, and has the terminal while it runs.
480    /// The runner sets it for the one task a run is for, when that task has a
481    /// single command whose output goes straight to the terminal.
482    pub interactive: bool,
483    /// Whether captured commands may run under a pseudo-terminal, which they
484    /// do when qk's output is a terminal: Nx's `usePty`. Commands whose
485    /// output is hashed set it to false, so keys never depend on a terminal.
486    pub pseudo_terminal: bool,
487    /// Nx's `tty`: whether qk's terminal is in raw input mode while the
488    /// interactive task reads it through a pseudo-terminal.
489    pub tty: bool,
490    /// What the commands run under, for `--sandbox`.
491    pub sandbox: Option<Confinement>,
492    pub recording: Option<Recording>,
493    /// Whether `recording` may key the task's result for `qk:ultracache`, as
494    /// `--record` gives it. A recording made only to audit the task does not.
495    pub recording_keys: bool,
496    /// Directories emptied before its commands run, which it keeps nothing
497    /// in between runs: its target's `qk:sandbox.scratch`.
498    pub scratch: Vec<Scratch>,
499    /// How the task's output is shown; the runner sets it per output style.
500    pub display: Display,
501}
502
503/// The dotenv files Nx loads for a task, most specific first: in the
504/// project root, then the workspace root, for the target and configuration
505/// (`.env.<target>.<configuration>`, `.env.<configuration>`, `.env.<target>`),
506/// then `.env.local`, `.local.env` and `.env`, each also in its `.local` and
507/// `.<name>.env` spellings. An atomized target also reads the files of the
508/// target it was split from, after its own of the same kind.
509fn dotenv_files(
510    project_root: &str,
511    target: &str,
512    configuration: Option<&str>,
513    non_atomized: Option<&str>,
514) -> Vec<String> {
515    let mut identifiers = Vec::new();
516    if let Some(configuration) = configuration {
517        identifiers.push(format!("{target}.{configuration}"));
518        if let Some(non_atomized) = non_atomized {
519            identifiers.push(format!("{non_atomized}.{configuration}"));
520        }
521        identifiers.push(configuration.to_owned());
522    }
523    identifiers.push(target.to_owned());
524    if let Some(non_atomized) = non_atomized {
525        identifiers.push(non_atomized.to_owned());
526    }
527    identifiers.push(String::new());
528    let variants = |identifier: &str, root: &str| -> Vec<String> {
529        let prefix = if root.is_empty() || root == "." {
530            String::new()
531        } else {
532            format!("{root}/")
533        };
534        if identifier.is_empty() {
535            [".env.local", ".local.env", ".env"]
536                .map(|name| format!("{prefix}{name}"))
537                .to_vec()
538        } else {
539            [
540                format!(".env.{identifier}.local"),
541                format!(".env.{identifier}"),
542                format!(".{identifier}.local.env"),
543                format!(".{identifier}.env"),
544            ]
545            .map(|name| format!("{prefix}{name}"))
546            .to_vec()
547        }
548    };
549    let mut files = Vec::new();
550    for root in [project_root, ""] {
551        for identifier in &identifiers {
552            // Colons in target names cannot name regular dotenv files on Windows.
553            #[cfg(windows)]
554            if identifier.contains(':') {
555                continue;
556            }
557            for file in variants(identifier, root) {
558                if !files.contains(&file) {
559                    files.push(file);
560                }
561            }
562        }
563    }
564    files
565}
566
567/// The target whose dotenv files a task reads, and the target it was
568/// atomized from, as Nx's getOwnerTargetForTask finds them: through the
569/// project's `metadata.targetGroups`, a target of the task's group that
570/// names a `metadata.nonAtomizedTarget`. Unlike Nx, a task that names one
571/// itself is its own owner, so each atomized target reads its own files
572/// rather than the first of its group's.
573fn owner_target<'a>(
574    project: &'a qk_config::Project,
575    target: &'a str,
576) -> (&'a str, Option<&'a str>) {
577    let non_atomized = |definition: &'a qk_config::Target| {
578        definition
579            .extra
580            .get("metadata")
581            .and_then(|metadata| metadata.get("nonAtomizedTarget"))
582            .and_then(Value::as_str)
583    };
584    if let Some((name, definition)) = project.targets.get_key_value(target)
585        && let Some(non_atomized) = non_atomized(definition)
586    {
587        return (name, Some(non_atomized));
588    }
589    let groups = project
590        .extra
591        .get("metadata")
592        .and_then(|metadata| metadata.get("targetGroups"))
593        .and_then(Value::as_object);
594    for group in groups.into_iter().flat_map(|groups| groups.values()) {
595        let members: Vec<&str> = group
596            .as_array()
597            .into_iter()
598            .flatten()
599            .filter_map(Value::as_str)
600            .collect();
601        if !members.contains(&target) {
602            continue;
603        }
604        for member in members {
605            if let Some((name, definition)) = project.targets.get_key_value(member)
606                && let Some(non_atomized) = non_atomized(definition)
607            {
608                return (name, Some(non_atomized));
609            }
610        }
611    }
612    (target, None)
613}
614
615/// The variable naming the tasks a task runs inside, one invocation a line.
616pub const INVOCATIONS: &str = "QK_TASK_INVOCATIONS";
617
618/// A task as an invocation: its id and arguments, which a task running inside
619/// it repeats only by recursing, as Nx's task invocation tracker tells it.
620pub fn invocation(task: &Task) -> String {
621    let mut text = task.id.clone();
622    for arg in &task.args {
623        text.push('\u{1f}');
624        text.push_str(arg);
625    }
626    text.replace('\n', " ")
627}
628
629/// The tasks this process runs inside, from [`INVOCATIONS`].
630pub fn enclosing_invocations() -> Vec<String> {
631    std::env::var(INVOCATIONS)
632        .map(|chain| chain.lines().map(str::to_owned).collect())
633        .unwrap_or_default()
634}
635
636/// The root dotenv files Nx loads into its own process at startup, in the
637/// order they take precedence.
638pub const ROOT_DOTENV_FILES: [&str; 3] = [".local.env", ".env.local", ".env"];
639
640/// A task's environment as Nx builds it: the process environment, without
641/// what the root dotenv files set in it at startup, then the task's dotenv
642/// files, loaded and expanded together, then the variables Nx sets for
643/// every task. A target's `env` applies on top.
644fn task_environment(
645    workspace: &Workspace,
646    task: &Task,
647    base_env: &BTreeMap<OsString, OsString>,
648) -> Result<(BTreeMap<OsString, OsString>, BTreeMap<OsString, OsString>)> {
649    let mut env = base_env.clone();
650    let project = &workspace.projects[&task.project];
651    if loads_dotenv(base_env) {
652        // As Nx's unloadDotEnvFiles: a variable whose value is a root file's
653        // goes, so the task's own files decide it. Nx unloads them even when
654        // it did not load them; qk keeps what the shell set then.
655        for file in [".env", ".local.env", ".env.local"] {
656            for (name, value) in dotenv::read_alone(&workspace.root.join(file))? {
657                if env.get(&name) == Some(&value) {
658                    env.remove(&name);
659                }
660            }
661        }
662        let (target, non_atomized) = owner_target(project, &task.target);
663        let files: Vec<PathBuf> = dotenv_files(
664            &project.root,
665            target,
666            task.configuration.as_deref(),
667            non_atomized,
668        )
669        .into_iter()
670        .map(|file| workspace.root.join(file))
671        .collect();
672        dotenv::load(&files, &mut env)?;
673    }
674    let runtime_env = env.clone();
675    // Nx's own variables from an enclosing run, which describe that run.
676    for name in [
677        "NX_SET_CLI",
678        "NX_ORIGINAL_FORCE_COLOR",
679        "NX_TERMINAL_OUTPUT_PATH",
680        "NX_STREAM_OUTPUT",
681        "NX_PREFIX_OUTPUT",
682        // A task of an enclosing run's signal socket and reported address:
683        // a task here neither signals for that task nor reads its address.
684        ready_signal::SOCKET_ENV,
685        ready_signal::TASK_ENV,
686        ready_signal::URL_ENV,
687    ] {
688        env.remove(std::ffi::OsStr::new(name));
689    }
690    let mut set = |name: &str, value: OsString| {
691        env.insert(name.into(), value);
692    };
693    set("NX_WORKSPACE_ROOT", workspace.root.clone().into_os_string());
694    set("NX_TASK_TARGET_PROJECT", task.project.clone().into());
695    set("NX_TASK_TARGET_TARGET", task.target.clone().into());
696    set("LERNA_PACKAGE_NAME", task.project.clone().into());
697    set("NX_TUI", "false".into());
698    // The qk and Nx processes this task runs inside, outermost first.
699    let ancestors = match base_env.get(std::ffi::OsStr::new("NX_INVOCATION_ANCESTOR_PIDS")) {
700        Some(inherited) if !inherited.is_empty() => {
701            format!("{},{}", inherited.to_string_lossy(), std::process::id())
702        }
703        _ => std::process::id().to_string(),
704    };
705    set("NX_INVOCATION_ANCESTOR_PIDS", ancestors.into());
706    // The tasks this one runs inside, outermost first, so a run it starts can
707    // tell when it would run one of them again.
708    let mut chain = base_env
709        .get(std::ffi::OsStr::new(INVOCATIONS))
710        .map(|chain| chain.to_string_lossy().into_owned())
711        .unwrap_or_default();
712    if !chain.is_empty() {
713        chain.push('\n');
714    }
715    chain.push_str(&invocation(task));
716    set(INVOCATIONS, chain.into());
717    if task.target == "test" {
718        set("NX_TERMINAL_CAPTURE_STDERR", "true".into());
719    }
720    match &task.configuration {
721        Some(configuration) => set("NX_TASK_TARGET_CONFIGURATION", configuration.clone().into()),
722        None => {
723            env.remove(std::ffi::OsStr::new("NX_TASK_TARGET_CONFIGURATION"));
724        }
725    }
726    Ok((env, runtime_env))
727}
728
729/// Whether dotenv files load: as in Nx, `NX_LOAD_DOT_ENV_FILES=false` turns
730/// them all off.
731fn loads_dotenv(env: &BTreeMap<OsString, OsString>) -> bool {
732    env.get(std::ffi::OsStr::new("NX_LOAD_DOT_ENV_FILES"))
733        .is_none_or(|value| value != "false")
734}
735
736/// The process environment with the root dotenv files loaded into it, as
737/// Nx loads them into its own process at startup, unless
738/// `NX_LOAD_DOT_ENV_FILES=false`. Variables already set win.
739pub fn environment(workspace: &Path) -> Result<BTreeMap<OsString, OsString>> {
740    let mut values: BTreeMap<OsString, OsString> = std::env::vars_os().collect();
741    if loads_dotenv(&values) {
742        let files: Vec<PathBuf> = ROOT_DOTENV_FILES
743            .iter()
744            .map(|file| workspace.join(file))
745            .collect();
746        dotenv::load(&files, &mut values)?;
747    }
748    Ok(values)
749}
750
751pub fn prepare(
752    workspace: &Workspace,
753    task: &Task,
754    base_env: &BTreeMap<OsString, OsString>,
755) -> Result<PreparedTask> {
756    let definition = &task.definition;
757    let executor = definition
758        .executor
759        .as_deref()
760        .context("target has no executor or command")?;
761    let project = &workspace.projects[&task.project];
762    let run_commands = executor == "nx:run-commands";
763    if !matches!(executor, "nx:run-commands" | "nx:run-script" | "nx:noop") {
764        bail!("unsupported executor {executor:?}");
765    }
766    // As in Nx, run-script reads only `script`, and noop nothing.
767    let options = if run_commands {
768        run_commands_overrides(&definition.options, &task.args, project)?
769    } else {
770        definition.options.clone()
771    };
772    let options = &options;
773    let empty = Map::new();
774    let mut interpolation = if run_commands {
775        interpolate::Interpolation::new(&task.args, options)?
776    } else {
777        interpolate::Interpolation::new(&[], &empty)?
778    };
779    let (mut env, mut runtime_env) = task_environment(workspace, task, base_env)?;
780    // As in Nx 24, `envFile` sets what the task's environment, dotenv files
781    // included, leaves unset or empty, beneath `env`; and the task's
782    // environment says whether it loads.
783    if let Some(file) = options.get("envFile").filter(|_| run_commands) {
784        let file = file.as_str().context("envFile must be a string")?;
785        let path = workspace.root.join(file);
786        if loads_dotenv(&runtime_env) {
787            if !path.is_file() {
788                bail!("envFile {file} does not exist");
789            }
790            dotenv::load(std::slice::from_ref(&path), &mut runtime_env)?;
791            dotenv::load(std::slice::from_ref(&path), &mut env)?;
792        }
793    }
794    // A target-level `env` is a qk extension, for every executor.
795    let option_env = options.get("env").filter(|_| run_commands);
796    for values in [definition.extra.get("env"), option_env]
797        .into_iter()
798        .flatten()
799    {
800        for (name, value) in values.as_object().context("env must be an object")? {
801            if name.is_empty() || name.contains(['=', '\0']) {
802                bail!("invalid environment variable name");
803            }
804            let value: OsString = value
805                .as_str()
806                .context("environment values must be strings")?
807                .into();
808            let name = environment_key(&env, name.into());
809            runtime_env.insert(name.clone(), value.clone());
810            env.insert(name, value);
811        }
812    }
813    interpolation.set_env(&env);
814    let mut execution = BTreeMap::new();
815    execution.insert(
816        "FORCE_COLOR".into(),
817        env.get(std::ffi::OsStr::new("FORCE_COLOR"))
818            .cloned()
819            .unwrap_or_else(|| "true".into()),
820    );
821    if run_commands && boolean(options.get("color"), false, "color")? {
822        execution.insert("FORCE_COLOR".into(), "true".into());
823    }
824    // run-commands runs where `cwd` says, the workspace root by default, and
825    // run-script in the project root.
826    let cwd = match options.get("cwd").filter(|_| run_commands) {
827        Some(cwd) => cwd.as_str().context("cwd must be a string")?,
828        None if run_commands => ".",
829        None => &project.root,
830    };
831    let cwd = workspace.root.join(cwd);
832    let mut resolved = PathBuf::new();
833    for part in cwd.components() {
834        match part {
835            std::path::Component::ParentDir => {
836                resolved.pop();
837            }
838            std::path::Component::CurDir => {}
839            _ => resolved.push(part.as_os_str()),
840        }
841    }
842    // Expose local binaries even for plain run-commands targets.
843    let mut paths = Vec::new();
844    for directory in resolved.ancestors() {
845        paths.push(directory.join("node_modules/.bin"));
846    }
847    let node_path = NodePath::new(&workspace.root, base_env, paths.len());
848    let path_key = path_key(&env);
849    if let Some(path) = env.get(&path_key) {
850        paths.extend(std::env::split_paths(path));
851    }
852    let path = std::env::join_paths(paths).context("cannot construct task PATH")?;
853    env.insert(path_key.clone(), path.clone());
854    runtime_env.insert(path_key, path);
855    let ready_when: Vec<String> = match options.get("readyWhen").filter(|_| run_commands) {
856        None | Some(Value::Null) => Vec::new(),
857        Some(Value::String(text)) => vec![text.clone()],
858        Some(value) => serde_json::from_value(value.clone())
859            .context("readyWhen must be a string or an array of strings")?,
860    };
861    // As in Nx, a single `command` runs in parallel exactly when it has
862    // `readyWhen`, whatever `parallel` says.
863    let parallel = if run_commands && options.get("command").is_some_and(merge_truthy) {
864        !ready_when.is_empty()
865    } else if run_commands {
866        boolean(options.get("parallel"), true, "parallel")?
867    } else {
868        true
869    };
870    let (mut pseudo_terminal, tty) = if run_commands {
871        (
872            boolean(options.get("usePty"), true, "usePty")?
873                && base_env
874                    .get(std::ffi::OsStr::new("NX_NATIVE_COMMAND_RUNNER"))
875                    .is_none_or(|value| value != "false"),
876            boolean(options.get("tty"), true, "tty")?,
877        )
878    } else {
879        (true, true)
880    };
881    let mut decorations = Vec::new();
882    if !ready_when.is_empty() && !parallel {
883        bail!("readyWhen can only be used when parallel is true");
884    }
885    let mut commands = Vec::new();
886    match executor {
887        "nx:noop" => {}
888        "nx:run-script" => {
889            let script = options
890                .get("script")
891                .and_then(Value::as_str)
892                .context("run-script requires options.script")?;
893            let package = workspace
894                .packages
895                .get(&task.project)
896                .context("run-script requires a package.json")?;
897            if !package.scripts.contains_key(script) {
898                bail!("package.json has no script {script:?}");
899            }
900            let manager = package_manager(workspace)?;
901            // Nx joins the arguments as they are; qk quotes each, so that
902            // one with a space or a shell character stays one argument.
903            let args = interpolate::quoted_arguments(&task.args)?;
904            let script = interpolate::quote(script)?;
905            // Nx's `run` for each: npm and bun need `--` before arguments,
906            // and yarn runs a script by name.
907            let mut command = match manager {
908                "yarn" => format!("yarn {script}"),
909                manager => format!("{manager} run {script}"),
910            };
911            if !args.is_empty() {
912                if matches!(manager, "npm" | "bun") {
913                    command.push_str(" --");
914                }
915                command.push(' ');
916                command.push_str(&args);
917            }
918            commands.push(command);
919        }
920        "nx:run-commands" => {
921            let forward = boolean(options.get("forwardAllArgs"), true, "forwardAllArgs")?;
922            // As in Nx, `command` wins over `commands`.
923            if let Some(command) = options
924                .get("command")
925                .filter(|command| merge_truthy(command))
926            {
927                // Nx also takes an array, joined with spaces.
928                let command = match command {
929                    Value::String(command) => command.clone(),
930                    Value::Array(words) => words
931                        .iter()
932                        .map(|word| word.as_str().context("command words must be strings"))
933                        .collect::<Result<Vec<_>>>()?
934                        .join(" "),
935                    _ => bail!("command must be a string or an array of strings"),
936                };
937                commands.push(interpolation.command(&command, forward)?);
938            } else if let Some(values) = options.get("commands") {
939                for value in values.as_array().context("commands must be an array")? {
940                    let (command, forward) = if let Some(command) = value.as_str() {
941                        (command, forward)
942                    } else {
943                        let value = value
944                            .as_object()
945                            .context("commands entries must be strings or objects")?;
946                        for key in value.keys() {
947                            if !matches!(
948                                key.as_str(),
949                                "command"
950                                    | "forwardAllArgs"
951                                    | "prefix"
952                                    | "prefixColor"
953                                    | "color"
954                                    | "bgColor"
955                                    | "description"
956                            ) {
957                                bail!("unsupported command entry field {key:?}");
958                            }
959                        }
960                        let field =
961                            |name: &str, background: Option<bool>| -> Result<Option<String>> {
962                                let Some(value) = value.get(name) else {
963                                    return Ok(None);
964                                };
965                                let text = value.as_str().with_context(|| {
966                                    format!("command entry {name} must be a string")
967                                })?;
968                                if let Some(background) = background
969                                    && !Decoration::valid(text, background)
970                                {
971                                    bail!("command entry {name} {text:?} is not a colour");
972                                }
973                                Ok(Some(text.to_owned()))
974                            };
975                        let decoration = Decoration {
976                            prefix: field("prefix", None)?,
977                            prefix_color: field("prefixColor", Some(false))?,
978                            color: field("color", Some(false))?,
979                            bg_color: field("bgColor", Some(true))?,
980                        };
981                        if !decoration.is_empty() && !parallel {
982                            bail!(
983                                "prefix, prefixColor, color and bgColor can only be set when parallel is true"
984                            );
985                        }
986                        decorations.resize(commands.len(), Decoration::default());
987                        decorations.push(decoration);
988                        (
989                            value
990                                .get("command")
991                                .and_then(Value::as_str)
992                                .context("command entry requires a command string")?,
993                            boolean(value.get("forwardAllArgs"), forward, "forwardAllArgs")?,
994                        )
995                    };
996                    commands.push(interpolation.command(command, forward)?);
997                }
998            } else {
999                bail!("run-commands requires command or commands");
1000            }
1001            // An empty `commands` succeeds without running anything, as in Nx.
1002            if commands.iter().any(|command| command.trim().is_empty()) {
1003                bail!("run-commands requires nonempty commands");
1004            }
1005        }
1006        _ => bail!("unsupported executor {executor:?}"),
1007    }
1008    // As in Nx, commands run side by side share qk's output through pipes.
1009    pseudo_terminal &= commands.len() == 1 || !parallel;
1010    let ready_inputs = ready_inputs(workspace, task, options)?.map(std::sync::Arc::new);
1011    let mut prepared = PreparedTask {
1012        id: task.id.clone(),
1013        commands,
1014        parallel,
1015        cwd,
1016        env,
1017        runtime_env,
1018        node_path: Some(node_path),
1019        execution,
1020        ready_when,
1021        ready: Default::default(),
1022        ready_check: None,
1023        ready_inputs,
1024        verdict: Default::default(),
1025        processes: Default::default(),
1026        decorations,
1027        interactive: false,
1028        pseudo_terminal,
1029        tty,
1030        sandbox: None,
1031        recording: None,
1032        recording_keys: false,
1033        scratch: Vec::new(),
1034        display: Display::default(),
1035    };
1036    prepared.ready_check = prepared.resolve_ready_check(None);
1037    Ok(prepared)
1038}
1039
1040/// A continuous target's `readyWhen` and what its placeholders read but the
1041/// environment: the options the task runs with, its arguments applied.
1042#[derive(Debug)]
1043pub struct ReadyInputs {
1044    config: Value,
1045    workspace_root: String,
1046    project_root: String,
1047    project_name: String,
1048    options: Map<String, Value>,
1049    /// `options` as a command check reads them, in `NX_TASK_OPTIONS`.
1050    options_json: String,
1051}
1052
1053/// A continuous target's `readyWhen`, as Nx 24 reads it. `null` turns off one
1054/// a target default sets. A target that is not continuous has none.
1055fn ready_inputs(
1056    workspace: &Workspace,
1057    task: &Task,
1058    options: &Map<String, Value>,
1059) -> Result<Option<ReadyInputs>> {
1060    let Some(ready_when) = task
1061        .definition
1062        .extra
1063        .get("readyWhen")
1064        .filter(|value| !value.is_null() && task.definition.continuous == Some(true))
1065    else {
1066        return Ok(None);
1067    };
1068    let mut options = options.clone();
1069    let project = &workspace.projects[&task.project];
1070    for (key, value) in interpolate::yargs(&task.args, false, false) {
1071        if key == "_" {
1072            continue;
1073        }
1074        // As in Nx, an argument's tokens resolve for the task's project.
1075        let value = match value {
1076            Value::String(text) => Value::String(qk_config::interpolate_override(&text, project)?),
1077            value => value,
1078        };
1079        if key.contains('-') {
1080            options.insert(interpolate::camel_case(&key), value.clone());
1081        }
1082        options.insert(key, value);
1083    }
1084    Ok(Some(ReadyInputs {
1085        config: ready_when.clone(),
1086        workspace_root: workspace
1087            .root
1088            .to_str()
1089            .context("workspace path must be UTF-8")?
1090            .to_owned(),
1091        project_root: project.root.clone(),
1092        project_name: task.project.clone(),
1093        options_json: serde_json::to_string(&options)?,
1094        options,
1095    }))
1096}
1097
1098impl PreparedTask {
1099    /// The task's own `readyWhen`, resolved against its options and the
1100    /// environment it runs with, with `url` as `NX_READY_URL`. As in Nx
1101    /// 24.0.1, a task waiting for a server that reports its address resolves
1102    /// it once that server is ready, so its placeholders and a command check
1103    /// read the address.
1104    pub fn resolve_ready_check(
1105        &self,
1106        url: Option<&str>,
1107    ) -> Option<std::result::Result<readiness::ReadyCheck, String>> {
1108        let inputs = self.ready_inputs.as_deref()?;
1109        let mut full_env = self.env.clone();
1110        if let Some(url) = url {
1111            full_env.insert(ready_signal::URL_ENV.into(), url.into());
1112        }
1113        let env: BTreeMap<String, String> = full_env
1114            .iter()
1115            .filter_map(|(name, value)| {
1116                Some((name.to_str()?.to_owned(), value.to_str()?.to_owned()))
1117            })
1118            .collect();
1119        let scope = readiness::Scope {
1120            workspace_root: &inputs.workspace_root,
1121            project_root: &inputs.project_root,
1122            project_name: &inputs.project_name,
1123            options: &inputs.options,
1124            env: &env,
1125        };
1126        let mut check = readiness::resolve(&self.id, &inputs.config, &scope);
1127        // As in Nx 24, a command check runs as the task's commands do, and
1128        // reads the options they run with.
1129        if let Ok(Some(readiness::ReadyCheck {
1130            probe: readiness::Probe::Command { env: probe_env, .. },
1131            ..
1132        })) = &mut check
1133        {
1134            *probe_env = full_env;
1135            probe_env.insert("NX_TASK_OPTIONS".into(), inputs.options_json.clone().into());
1136        }
1137        check.map_err(|error| format!("{error:#}")).transpose()
1138    }
1139}
1140
1141/// The run-commands options Nx reads, which a task's arguments may set too.
1142const OPTION_FLAGS: &[&str] = &[
1143    "command",
1144    "commands",
1145    "color",
1146    "parallel",
1147    "readyWhen",
1148    "cwd",
1149    "args",
1150    "envFile",
1151    "__unparsed__",
1152    "env",
1153    "usePty",
1154    "streamOutput",
1155    "verbose",
1156    "forwardAllArgs",
1157    "tty",
1158];
1159
1160/// The target's options with the task's arguments that name a run-commands
1161/// option applied. As in Nx, `--cwd=dir`, `--no-parallel` or
1162/// `--env.NAME=value` set the option rather than reaching the command.
1163fn run_commands_overrides(
1164    options: &serde_json::Map<String, Value>,
1165    args: &[String],
1166    project: &qk_config::Project,
1167) -> Result<serde_json::Map<String, Value>> {
1168    const BOOLEAN: &[&str] = &[
1169        "color",
1170        "parallel",
1171        "usePty",
1172        "streamOutput",
1173        "verbose",
1174        "forwardAllArgs",
1175        "tty",
1176    ];
1177    let mut options = options.clone();
1178    let mut ready = Vec::new();
1179    let mut args = args.iter().peekable();
1180    while let Some(arg) = args.next() {
1181        let Some(flag) = arg.strip_prefix("--") else {
1182            continue;
1183        };
1184        let (key, value) = match flag.split_once('=') {
1185            Some((key, value)) => (key, Some(value.to_owned())),
1186            None => (flag, None),
1187        };
1188        let (key, negated) = match key.strip_prefix("no-") {
1189            Some(key) if OPTION_FLAGS.contains(&key) => (key, true),
1190            _ => (key, false),
1191        };
1192        let (base, field) = match key.split_once('.') {
1193            Some((base, field)) => (base, Some(field)),
1194            None => (key, None),
1195        };
1196        if !OPTION_FLAGS.contains(&base) {
1197            continue;
1198        }
1199        let boolean = BOOLEAN.contains(&base);
1200        let value = match value {
1201            Some(value) => value,
1202            None if negated => "false".into(),
1203            None if !boolean && args.peek().is_some_and(|next| !next.starts_with('-')) => {
1204                args.next().unwrap().clone()
1205            }
1206            None => "true".into(),
1207        };
1208        let value = if boolean {
1209            Value::Bool(value != "false")
1210        } else {
1211            // As in Nx, an argument's tokens resolve for the task's project.
1212            Value::String(qk_config::interpolate_override(&value, project)?)
1213        };
1214        match (base, field) {
1215            ("env", Some(name)) => {
1216                let env = options
1217                    .entry("env")
1218                    .or_insert_with(|| Value::Object(Default::default()));
1219                if let Value::Object(env) = env {
1220                    env.insert(name.to_owned(), value);
1221                }
1222            }
1223            ("readyWhen", None) => ready.push(value),
1224            // Nx sets these itself, and a list cannot come from a flag.
1225            ("__unparsed__" | "commands", _) | (_, Some(_)) => {}
1226            (base, None) => {
1227                options.insert(base.to_owned(), value);
1228            }
1229        }
1230    }
1231    if !ready.is_empty() {
1232        let ready = if ready.len() == 1 {
1233            ready.pop().unwrap()
1234        } else {
1235            Value::Array(ready)
1236        };
1237        options.insert("readyWhen".into(), ready);
1238    }
1239    Ok(options)
1240}
1241
1242/// JavaScript truthiness, which decides what Nx reads as set.
1243fn merge_truthy(value: &Value) -> bool {
1244    match value {
1245        Value::Null => false,
1246        Value::Bool(flag) => *flag,
1247        Value::Number(number) => number.as_f64().is_some_and(|number| number != 0.0),
1248        Value::String(text) => !text.is_empty(),
1249        Value::Array(_) | Value::Object(_) => true,
1250    }
1251}
1252
1253fn boolean(value: Option<&Value>, default: bool, name: &str) -> Result<bool> {
1254    value
1255        .map(|value| {
1256            value
1257                .as_bool()
1258                .with_context(|| format!("{name} must be boolean"))
1259        })
1260        .transpose()
1261        .map(|value| value.unwrap_or(default))
1262}
1263
1264fn package_manager(workspace: &Workspace) -> Result<&'static str> {
1265    workspace.package_manager_name()
1266}
1267
1268#[cfg(all(test, windows))]
1269mod dotenv_tests {
1270    #[test]
1271    fn colon_targets_keep_configuration_and_generic_dotenv_files() {
1272        let files = super::dotenv_files("app", "test:unit:watch", Some("ci"), None);
1273        assert!(files.iter().all(|file| !file.contains(':')));
1274        assert!(files.contains(&"app/.env.ci".to_owned()));
1275        assert!(files.contains(&"app/.env".to_owned()));
1276        assert!(files.contains(&".env.ci".to_owned()));
1277        assert!(files.contains(&".env".to_owned()));
1278    }
1279}