path query: parallel per-file execution (path-cli 0.17.0)

CHANGELOG.md

@@ -2,6 +2,36 @@
 
 All notable changes to the Toolpath workspace are documented here.
 
+## Parallel `path query` execution — 2026-08-06
+
+`path query` now evaluates cache documents on a thread pool. For the
+plans that treat every file independently (`PerFileStream`,
+`Decompose`), the whole per-file pipeline — read, parse, wrap, filter,
+render — runs on rayon workers, with only ordered output assembly on
+the main thread; `Slurp` plans parallelize the parse/wrap phase only
+(the merged jaq array is `Rc`-based and must be built on one thread).
+Output is byte-identical to the sequential scan — same ordering, same
+warnings, same error precedence — enforced by tests pinning the
+parallel building blocks against the sequential engine.
+
+Measured on a real 97-doc / 114 MB cache (M-series, medians of 5):
+streamed and decomposed queries drop from ~0.95 s to ~0.31 s (~3.1×),
+slurped queries improve ~1.2×. On an even 60-doc / 56 MB synthetic
+cache the same queries run ~4.7× faster (e.g. `length` 966 ms →
+206 ms).
+
+- **`path-cli`** (0.17.0):
+  - `query/mod.rs`: plan dispatch (`execute_plan`) and a chunked
+    parallel driver (`for_each_file`) that preserves selection order,
+    per-file warning order, and explicit-file error positions.
+  - `query/filter.rs`: per-file worker evaluation (`render_file`,
+    `partials_file` — partials cross threads as compact JSON bytes and
+    are reparsed on the consuming thread), a per-thread compiled-filter
+    cache, and `finish_decompose` shared by the sequential and parallel
+    drivers so the zero-file rule lives once.
+  - The emscripten (playground) build keeps the sequential engine —
+    no threads there.
+
 ## Projected Claude sessions are resumable again — 2026-07-30
 
 Two fixes found by live-resuming a projected session against the real

Cargo.toml

@@ -37,7 +37,7 @@ toolpath-github = { version = "0.6.0", path = "crates/toolpath-github" }
 toolpath-dot = { version = "0.5.0", path = "crates/toolpath-dot" }
 toolpath-md = { version = "0.7.0", path = "crates/toolpath-md" }
 toolpath-pi = { version = "0.6.1", path = "crates/toolpath-pi" }
-path-cli = { version = "0.16.1", path = "crates/path-cli" }
+path-cli = { version = "0.17.0", path = "crates/path-cli" }
 pathbase-client = { version = "0.2.0", path = "crates/pathbase-client" }
 
 reqwest = { version = "0.13", default-features = false, features = ["blocking", "json", "rustls"] }
@@ -54,6 +54,7 @@ similar = "2"
 tempfile = "3.15"
 insta = "1"
 rusqlite = { version = "0.32", features = ["bundled"] }
+rayon = "1.10"
 uuid = { version = "1", features = ["serde"] }
 
 [profile.wasm]

crates/path-cli/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "path-cli"
-version = "0.16.1"
+version = "0.17.0"
 edition.workspace = true
 license.workspace = true
 repository = "https://github.com/empathic/toolpath"
@@ -56,6 +56,7 @@ git2 = { workspace = true }
 reqwest = { workspace = true }
 tokio = { workspace = true }
 rusqlite = { workspace = true }
+rayon = { workspace = true }
 uuid = { workspace = true, features = ["v4"] }
 # Embedded fuzzy picker — used when external `fzf` isn't on PATH.
 # Disable default features to avoid pulling in skim's CLI deps

crates/path-cli/src/query/filter.rs

@@ -59,7 +59,6 @@ pub fn execute(
             eval_print(&main, merged, out, &pp, raw)?;
         }
         Plan::Decompose { reduce } => {
-            let reducer = compile(reduce)?;
             let mut partials: Vec<Val> = Vec::new();
             let mut saw_file = false;
             let mut emit = |val: Val| {
@@ -68,23 +67,109 @@ pub fn execute(
                 Ok(())
             };
             run_files(&mut emit)?;
-            if saw_file {
-                let merged: Val = partials.into_iter().collect();
-                eval_print(&reducer, merged, out, &pp, raw)?;
-            } else {
-                // No document contributed a partial: the decomposition
-                // identity `reduce(⋃ main(fᵢ)) == main(⋃ fᵢ)` degenerates to
-                // `main([])`. Run the *main* filter over an empty array so the
-                // answer matches slurp (`length` → 0, `sort_by|.[:N]` → []),
-                // not the reducer over `[]` (which would give null / error).
-                let empty: Val = std::iter::empty().collect();
-                eval_print(&main, empty, out, &pp, raw)?;
-            }
+            finish_decompose(main_src, reduce, partials, saw_file, compact, raw, out)?;
         }
     }
     Ok(())
 }
 
+/// Finish a `Decompose` plan over the gathered per-file partials — shared by
+/// the sequential and parallel drivers so the zero-file rule lives once.
+///
+/// With no document contributing a partial, the decomposition identity
+/// `reduce(⋃ main(fᵢ)) == main(⋃ fᵢ)` degenerates to `main([])`. Run the
+/// *main* filter over an empty array so the answer matches slurp (`length`
+/// → 0, `sort_by|.[:N]` → []), not the reducer over `[]` (which would give
+/// null / error).
+pub fn finish_decompose(
+    main_src: &str,
+    reduce_src: &str,
+    partials: Vec<Val>,
+    saw_file: bool,
+    compact: bool,
+    raw: bool,
+    out: &mut dyn Write,
+) -> Result<()> {
+    let pp = pretty(compact);
+    if saw_file {
+        let reducer = compile(reduce_src)?;
+        let merged: Val = partials.into_iter().collect();
+        eval_print(&reducer, merged, out, &pp, raw)
+    } else {
+        let main = compile(main_src)?;
+        let empty: Val = std::iter::empty().collect();
+        eval_print(&main, empty, out, &pp, raw)
+    }
+}
+
+/// Surface a filter's load/compile errors without running it. The parallel
+/// drivers call this before touching any file, preserving the sequential
+/// path's error precedence (a bad filter beats a missing `--id`).
+#[cfg(not(target_os = "emscripten"))]
+pub fn compile_check(code: &str) -> Result<()> {
+    compile(code).map(|_| ())
+}
+
+/// Compile `code` at most once per thread. Rayon workers are long-lived, so
+/// per-file evaluation pays one compile per pool thread, not per file.
+#[cfg(not(target_os = "emscripten"))]
+fn with_compiled<T>(code: &str, f: impl FnOnce(&Program) -> Result<T>) -> Result<T> {
+    use std::cell::RefCell;
+    thread_local! {
+        static CACHE: RefCell<Option<(String, Program)>> = const { RefCell::new(None) };
+    }
+    CACHE.with(|cache| {
+        let mut cache = cache.borrow_mut();
+        if !matches!(&*cache, Some((cached, _)) if cached == code) {
+            *cache = Some((code.to_string(), compile(code)?));
+        }
+        let (_, prog) = cache.as_ref().expect("just populated");
+        f(prog)
+    })
+}
+
+/// Evaluate `code` over one file's steps and render its outputs exactly as
+/// [`execute`]'s streaming printer would. Runs on worker threads.
+#[cfg(not(target_os = "emscripten"))]
+pub fn render_file(
+    code: &str,
+    steps: Vec<serde_json::Value>,
+    compact: bool,
+    raw: bool,
+) -> Result<Vec<u8>> {
+    with_compiled(code, |prog| {
+        let mut buf = Vec::new();
+        eval_print(prog, steps_to_val(steps)?, &mut buf, &pretty(compact), raw)?;
+        Ok(buf)
+    })
+}
+
+/// Evaluate `code` over one file's steps and pack the partial outputs into
+/// one compact JSON array. Jaq values are `Rc`-based and thread-local, so
+/// partials cross threads as bytes; [`unpack_partials`] reparses them on the
+/// consuming thread.
+#[cfg(not(target_os = "emscripten"))]
+pub fn partials_file(code: &str, steps: Vec<serde_json::Value>) -> Result<Vec<u8>> {
+    with_compiled(code, |prog| {
+        let vals = eval_collect(prog, steps_to_val(steps)?)?;
+        let arr: Val = vals.into_iter().collect();
+        let mut buf = Vec::new();
+        jaq_json::write::write(&mut buf, &pretty(true), 0, &arr)
+            .map_err(|e| anyhow!("internal: could not pack partials: {e}"))?;
+        Ok(buf)
+    })
+}
+
+/// Reparse one file's packed partials (the inverse of [`partials_file`]).
+#[cfg(not(target_os = "emscripten"))]
+pub fn unpack_partials(bytes: &[u8]) -> Result<Vec<Val>> {
+    match jaq_json::read::parse_single(bytes) {
+        Ok(Val::Arr(items)) => Ok(items.iter().cloned().collect()),
+        Ok(_) => Err(anyhow!("internal: packed partials were not an array")),
+        Err(e) => Err(anyhow!("internal: could not reparse partials: {e}")),
+    }
+}
+
 /// Convert one file's wrapped steps into a jaq array value. The byte buffer is
 /// per-file (bounded), so no whole-cache serialization is ever held.
 pub fn steps_to_val(steps: Vec<serde_json::Value>) -> Result<Val> {
@@ -402,4 +487,93 @@ mod tests {
         // filter.
         assert_slurps("map({id: .step.id}) | (sort_by(.id)) | .[:1]");
     }
+
+    // ── The parallel drivers' building blocks ─────────────────────────
+    //
+    // `render_file`/`partials_file` + `finish_decompose` are what the
+    // parallel per-file drivers run on worker threads. Their output must be
+    // byte-identical to the sequential engine, which is itself pinned to
+    // slurp above.
+
+    fn file_steps(f: &serde_json::Value) -> Vec<serde_json::Value> {
+        f.as_array().cloned().unwrap_or_default()
+    }
+
+    #[test]
+    fn render_file_concat_matches_sequential_stream() {
+        for (compact, raw) in [(true, false), (false, false), (true, true)] {
+            for code in [
+                ".[] | select(.dead_end)",
+                r#".[] | select(.step.actor | startswith("agent:")) | .step.id"#,
+                "map(select(.tokens > 40))",
+            ] {
+                let mut seq: Vec<u8> = Vec::new();
+                execute(&Plan::PerFileStream, code, compact, raw, &mut seq, |emit| {
+                    for f in &fixture() {
+                        emit(steps_to_val(file_steps(f))?)?;
+                    }
+                    Ok(())
+                })
+                .unwrap();
+
+                let mut par: Vec<u8> = Vec::new();
+                for f in &fixture() {
+                    par.extend(render_file(code, file_steps(f), compact, raw).unwrap());
+                }
+                assert_eq!(
+                    String::from_utf8(par).unwrap(),
+                    String::from_utf8(seq).unwrap(),
+                    "parallel render must equal sequential for `{code}` (compact={compact}, raw={raw})"
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn partials_roundtrip_matches_sequential_decompose() {
+        for code in [
+            "length",
+            "map(select(.dead_end)) | length",
+            "sort_by(-.tokens) | .[:2]",
+            "map({id: .step.id, t: .tokens})",
+        ] {
+            let plan = crate::query::plan::analyze(code);
+            let Plan::Decompose { reduce } = &plan else {
+                panic!("`{code}` should decompose");
+            };
+            let seq = run_with(&plan, code, &fixture());
+
+            let mut partials: Vec<Val> = Vec::new();
+            let mut saw_file = false;
+            for f in &fixture() {
+                saw_file = true;
+                let bytes = partials_file(code, file_steps(f)).unwrap();
+                partials.extend(unpack_partials(&bytes).unwrap());
+            }
+            let mut par: Vec<u8> = Vec::new();
+            finish_decompose(code, reduce, partials, saw_file, true, false, &mut par).unwrap();
+            assert_eq!(
+                String::from_utf8(par).unwrap(),
+                seq,
+                "parallel decompose must equal sequential for `{code}`"
+            );
+        }
+    }
+
+    #[test]
+    fn finish_decompose_zero_files_matches_slurp() {
+        for code in ["length", "sort_by(.tokens) | .[:2]"] {
+            let plan = crate::query::plan::analyze(code);
+            let Plan::Decompose { reduce } = &plan else {
+                panic!("`{code}` should decompose");
+            };
+            let mut par: Vec<u8> = Vec::new();
+            finish_decompose(code, reduce, Vec::new(), false, true, false, &mut par).unwrap();
+            let none: &[serde_json::Value] = &[];
+            assert_eq!(
+                String::from_utf8(par).unwrap(),
+                run_with(&Plan::Slurp, code, none)
+            );
+        }
+    }
 }