diff --git a/CHANGELOG.md b/CHANGELOG.md index 54cf353f..128900a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,54 @@ All notable changes to the Toolpath workspace are documented here. +## SQLite step index for `path query` — 2026-07-21 + +`path query` now keeps a disposable SQLite accelerator at +`$CONFIG_DIR/index.db`: every wrapped step as a row (exactly what the +jaq engine consumes), with generated columns over the hot fields and a +stat-level freshness gate per document. The cache files stay canonical +— delete the index and the next query rebuilds it lazily; a schema +version bump rebuilds it automatically. `TOOLPATH_QUERY_NO_INDEX=1` +bypasses it. + +What it accelerates (real 97-doc / 114 MB / 46k-step cache, medians; +"rayon" = 0.17.0's parallel scan): + +- **Absorbed counts** — a filter that is exactly `length` or + `map(select(P)) | length` with a recognized `P` becomes + `SELECT count(*)`: 0.90 s → **30 ms** (~30×; no step is parsed at + all). Recognized `P` atoms: `.dead_end` (and `| not`), + `.step.actor == "…"`, `.step.actor | startswith("…")`, + `.path.meta.source == "…"`, and `and`-conjunctions of those. +- **Predicated scans** — a leading `map(select(P))` / `.[] | select(P)` + prefilters rows in SQL before parsing: `map(select(.step.actor | + startswith("agent:"))) | length` 0.98 s → 169 ms (~6×; 310 ms under + rayon alone). +- Unrecognized/slurp queries are unchanged (they ride 0.17.0's + parallel scan); a predicated stream lands at parity with it + (row-fetch + per-row parse ≈ parallel whole-file parse). + +Correctness: recognition is exact (whole-predicate or nothing) and the +prefilter feeds the *unchanged* original filter, so pruned rows are +ones the filter's own first stage would drop — the integration suite +asserts index-served output is byte-identical to the no-index path +across every plan, staleness self-healing included. Content-scoped +queries (`--kind`/`--project`/`--parent-dir`) bypass the index. + +- **`path-cli`** (0.18.0): + - `query/index.rs`: the store — WAL, per-thread connections, + `(mtime_ns, size)` stamps, write-through from `cache::write_cached` + (sync/import keep the index warm), purge on `p cache rm`, orphan + pruning on full scans, version-gated self-rebuild. One-time build + cost on first query: ~1.7 s for the 114 MB cache; index size ~250 MB + (rows + hot-field indexes). + - `query/plan.rs`: `RowPredicate` — conservative recognizer for the + leading-`select` predicate and the absorbable-count shape, with an + in-code `matches` mirror used when a stale doc is reparsed. + - `TOOLPATH_QUERY_EXPLAIN=1` now also prints the index strategy + (`count absorbed into SQL (dead_end=true)`, `serving fresh docs, + prefilter …`, or `off`). + ## Parallel `path query` execution — 2026-07-21 `path query` now evaluates cache documents on a thread pool. For the diff --git a/CLAUDE.md b/CLAUDE.md index 8764e695..5089401e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -212,7 +212,7 @@ Tests live alongside the code (`#[cfg(test)] mod tests`), plus `path-cli` has in - `toolpath-cursor`: 78 unit + 8 integration round-trip + 1 real-DB sanity + 1 doc test (state.vscdb SQLite reader, bubble store + composer header parsing, content-addressed blob lookup, projector with full TOOL_TABLE coverage, JSONL transcript ingest in `examples/dump_fixture.rs`) - `toolpath-pi`: 133 unit + 26 integration + 5 doc tests (types, paths, error, reader, io, provider) - `toolpath-dot`: 30 unit + 2 doc tests (render, visual conventions, escaping) -- `path-cli`: 361 unit + 121 integration tests (import/export/cache, track sessions, merge, validate, roundtrip, render-md snapshots, deprecation aliases, pathbase HTTP mock-server tests, fzf-friendly TSV output, `path resume` orchestration with injectable `ExecStrategy`, `path query`/`path kind` jaq filters + kind-selector matching + step wrapping over a `$TOOLPATH_CONFIG_DIR` cache sandbox, streaming-planner recognition + streamed-output-equals-slurp equality checks, `p cache sync` incremental ingestion — stat fingerprints, refresh-overwrite, per-type filtering, failure tallying — over resolver-injected provider fixtures, and import/share manifest recording end-to-end). For an end-to-end check against a real Pathbase deployment, run `scripts/test-pathbase-live.sh ` — it does an anon round-trip in a sandboxed config dir and, if you're logged into that URL, an authed pathstash round-trip too. +- `path-cli`: 372 unit + 125 integration tests (import/export/cache, track sessions, merge, validate, roundtrip, render-md snapshots, deprecation aliases, pathbase HTTP mock-server tests, fzf-friendly TSV output, `path resume` orchestration with injectable `ExecStrategy`, `path query`/`path kind` jaq filters + kind-selector matching + step wrapping over a `$TOOLPATH_CONFIG_DIR` cache sandbox, streaming-planner recognition + streamed-output-equals-slurp equality checks, `p cache sync` incremental ingestion — stat fingerprints, refresh-overwrite, per-type filtering, failure tallying — over resolver-injected provider fixtures, and import/share manifest recording end-to-end). For an end-to-end check against a real Pathbase deployment, run `scripts/test-pathbase-live.sh ` — it does an anon round-trip in a sandboxed config dir and, if you're logged into that URL, an authed pathstash round-trip too. - `toolpath-cli`: 0 tests (it's a one-line `path_cli::run()` shim crate that exists only so `cargo install toolpath-cli` keeps installing the `path` binary) Validate example documents: `for f in examples/*.json; do cargo run -p path-cli -- p validate --input "$f"; done` @@ -281,6 +281,6 @@ Build the site after changes: `cd site && pnpm run build` (should produce 11 pag - Conversation metadata title field: `toolpath-claude::ConversationMetadata`, `toolpath-gemini::ConversationMetadata`, and `toolpath-pi::SessionMeta` all expose `first_user_message: Option` — the first non-empty user-prompt text. Populated cheaply during the metadata pass (single-pass for Claude/Gemini; one extra short read for Pi). Used by the picker UI but useful for any "list sessions by topic" surface. - `path share` is the one-shot equivalent of `path p import | path p export pathbase`. It probes installed agent harnesses (claude/gemini/codex/opencode/pi), aggregates their sessions into a single fzf picker, and ranks rows whose project (claude/gemini/pi) or recorded cwd (codex/opencode) canonicalizes to the current directory at the top. `--harness` narrows the picker to one provider; `--harness X --session Y` (and `--project P` for keyed providers) skips the picker entirely. Pathbase flags (`--url`, `--anon`, `--repo`, `--slug`, `--public`) match `path export pathbase`. By default the derived doc is written to the cache like `import` does; pass `--no-cache` to skip. When the manifest shows the picked session unchanged since its last sync (`sync::fresh_cache_id`: stamps match, doc present; the freshness stat targets that one artifact directly, no sibling enumeration), share uploads the cached doc directly instead of re-deriving. The cache ingests maximally (thinking always included), and uploads carry the same full derivation as local projection (`resume`, `p export `) — there is no egress stripping. - `path resume ` is the inverse of `path share`. It accepts a Pathbase URL, an `owner/repo/slug` shorthand, a local toolpath JSON file, or a cache id; resolves it (caching URL fetches under `~/.toolpath/documents/` unless `--no-cache`); validates that the document is a single agent-bearing `Path`; then opens an `fzf` harness picker (skipped with `--harness X`). The picker pre-selects the source harness inferred from `path.meta.source` (`claude-code`/`gemini-cli`/`codex`/`opencode`/`pi`) when it's installed. After picking, `path resume` projects the session into the harness's on-disk layout under the chosen working directory (default: shell cwd; override with `-C, --cwd P`) and `execvp`'s the harness's resume command (`claude -r ` / `gemini --resume ` / `codex resume ` / `opencode --session ` / `pi --session `). On Windows it spawns and waits, propagating the exit code. The exec is mockable via `cmd_resume::ExecStrategy` — production uses `RealExec`; integration tests use `RecordingExec` to capture the recipe without launching a real harness. -- `path query` does not load the whole cache into memory when it can avoid it. `crates/path-cli/src/query/plan.rs` parses the jaq filter into jaq's own AST (`jaq_core::load::parse::Term`) and classifies it into a `Plan`: `PerFileStream` (`.[] | g` element-wise work — run per document, print as you go), `Decompose { reduce }` (algebraic aggregations — run the whole filter per file, concatenate the per-file outputs, then run a derived combine: `map`→`add` (array concat), top-N `sort_by(k)|.[:N]`→`add | sort_by(k)|.[:N]`, `length`→`add` over exact integer counts), or `Slurp` (the always-correct whole-array fallback). Recognition is conservative — a non-distributive prefix like `unique`/`group_by` slurps, and so do scalar `add` (float sums re-associate across per-file partials), `min`/`max` (`[] | min == null` poisons the merge), and any unrecognized tail — so **the planner never changes an answer** — `crates/path-cli/src/query/filter.rs` tests assert streamed output equals slurp byte-for-byte. `filter::execute` compiles the filter once (jaq's compiled `Filter` is fully owned, so it's reused across files) and drives the plan; `mod.rs::stream_files` yields one document's wrapped steps at a time. Execution is parallel (`mod.rs::execute_plan`/`for_each_file`): `PerFileStream`/`Decompose` run the whole per-file pipeline (parse → wrap → filter → render/pack) on rayon workers in chunks — partials cross threads as compact JSON bytes since jaq `Val`s are `Rc`-based — while `Slurp` parallelizes only parse/wrap; output stays byte-identical to a sequential scan (ordering, warnings, error precedence), and the emscripten build stays fully sequential. `TOOLPATH_QUERY_EXPLAIN=1` prints the chosen plan to stderr. No user-facing flag — it's automatic. Tie-break caveat: a streamed top-N matches slurp's *ranking*, but boundary ties may resolve to different specific rows. +- `path query` does not load the whole cache into memory when it can avoid it. `crates/path-cli/src/query/plan.rs` parses the jaq filter into jaq's own AST (`jaq_core::load::parse::Term`) and classifies it into a `Plan`: `PerFileStream` (`.[] | g` element-wise work — run per document, print as you go), `Decompose { reduce }` (algebraic aggregations — run the whole filter per file, concatenate the per-file outputs, then run a derived combine: `map`→`add` (array concat), top-N `sort_by(k)|.[:N]`→`add | sort_by(k)|.[:N]`, `length`→`add` over exact integer counts), or `Slurp` (the always-correct whole-array fallback). Recognition is conservative — a non-distributive prefix like `unique`/`group_by` slurps, and so do scalar `add` (float sums re-associate across per-file partials), `min`/`max` (`[] | min == null` poisons the merge), and any unrecognized tail — so **the planner never changes an answer** — `crates/path-cli/src/query/filter.rs` tests assert streamed output equals slurp byte-for-byte. `filter::execute` compiles the filter once (jaq's compiled `Filter` is fully owned, so it's reused across files) and drives the plan; `mod.rs::stream_files` yields one document's wrapped steps at a time. Execution is parallel (`mod.rs::execute_plan`/`for_each_file`): `PerFileStream`/`Decompose` run the whole per-file pipeline (parse → wrap → filter → render/pack) on rayon workers in chunks — partials cross threads as compact JSON bytes since jaq `Val`s are `Rc`-based — while `Slurp` parallelizes only parse/wrap; output stays byte-identical to a sequential scan (ordering, warnings, error precedence), and the emscripten build stays fully sequential. A disposable SQLite step index at `$CONFIG_DIR/index.db` (`query/index.rs`; wrapped-step rows + generated hot-field columns, stat-gated per doc, write-through from `cache::write_cached`, purged by `p cache rm`, rebuilt on schema-version bump) accelerates two shapes: a filter that is exactly `length`/`map(select(P))|length` with a recognized `P` (`plan.rs::RowPredicate`: `.dead_end`, actor eq/startswith, `.path.meta.source` eq, `and`-conjunctions) collapses to `SELECT count(*)` (~30×, sub-50 ms on a 114 MB cache), and a recognized leading `select` prefilters rows in SQL before parsing. Recognition is whole-predicate-or-nothing and the unchanged original filter still runs, so answers never change (integration tests assert byte-equality vs `TOOLPATH_QUERY_NO_INDEX=1`); content-scoped queries (`--kind`/`--project`/`--parent-dir`) bypass the index. `TOOLPATH_QUERY_EXPLAIN=1` prints the chosen plan and index strategy to stderr. No user-facing flag — it's automatic. Tie-break caveat: a streamed top-N matches slurp's *ranking*, but boundary ties may resolve to different specific rows. - Cache sync: `path p cache sync [types…]` (`crates/path-cli/src/artifact.rs`: `ArtifactType` + `ArtifactRef` + the stamp helpers; `sync/engine.rs`: manifest + ingestion loop, no UI — it reports through a `SyncObserver` trait, `&mut ()` for a silent sync; `sync/sources.rs`: an `ArtifactSource` trait — enumerate / stamp / derive / peek-dir / scope-match — with one impl per provider, so the engine never matches on artifact type; `cmd_cache.rs`: the stderr progress line + summary) incrementally ingests artifacts into the cache — no args syncs every artifact type. Change detection is **stat-level**: each artifact is enumerated as an `ArtifactRef` whose fingerprint is the source file's mtime + size (claude: the *whole session chain* — max segment mtime + summed segment sizes via `claude_chain_stamp`, because Claude Code rotates to a new file on continuation while the chain keeps its oldest segment's id, so appends land in the newest file, not the head; the chain comes from the same cached index `list_conversations` builds; codex: rollout file, id from the stem's trailing UUID; pi: session file, id from a one-line header peek; copilot: `session-state//events.jsonl`, pure read-dir + stat) or the DB row's updated-at (opencode: header-only `SELECT time_updated`; cursor: composer headers' `lastUpdatedAt`, bubble-less drafts skipped, workspace-less composers *included* unlike `share`). Gemini enumerates via `PathResolver::list_session_entries` (`toolpath-gemini` 0.6.1), whose identity peek is bounded to the first 4 KiB of a main file. Deciding "nothing changed" reads no session bodies — a no-op sync is milliseconds. Changed/new artifacts derive through the same provider managers (each source calls the `derive_*_session_with` helpers in `derive.rs`). Manifest at `~/.toolpath/manifest.json`: artifact type → artifact id → `{path?, cache_id, modified?, size?, synced_at}`; atomic temp+rename writes, `0600`, checkpointed every 10 writes (interruption-safe: a killed run keeps nearly everything it derived, and derives run newest-first so partial progress covers the sessions that matter most); writers serialize on an advisory lock (`manifest.json.lock`) and every write is a locked read-merge-save — checkpoints merge only the records the run wrote — so concurrent invocations (query auto-syncs, imports) union their records instead of clobbering each other. Pending work reports progress on stderr (`\r`-updating ` done/total` on a TTY, a plain line every 25 items otherwise; no-op syncs stay silent). Sync always writes the cache with force — refresh semantics — and never deletes: artifacts removed upstream keep their cache docs and manifest records (archive, not mirror). Derivation failures warn and tally, they don't abort. A record's `cache_id` is *optional*: a record without one is "known, not materialized" — created when a `--parent-dir` constraint excluded a peeked artifact, or when `p cache rm` evicts a doc (rm downgrades the record; the next in-scope sync re-materializes it, and sync also verifies the doc file actually exists before skipping, so even out-of-band deletions self-heal). `--parent-dir ` / `-d` on both `p cache sync` and `path query` restricts ingestion to artifacts under that directory (subtree): path-keyed providers prune whole projects before enumerating (claude compares in *slug space* — its dir slugs are lossy, `/`/`_`/`.` all became `-`), cwd-keyed ones check the directory their cheap headers carry, and codex/copilot — whose cwd lives inside the session file — get a one-line peek only when new/changed, memoized into the record's `path`. The stat gate always runs first: unchanged+cached artifacts skip before any scope check. Out-of-scope work is tallied separately (`N out of scope`) and never touches a materialized record's stamp. Claude derives leave `DeriveConfig.project_path` unset so `path.base` comes from the session's own recorded cwd rather than the lossy slug. `path query` runs this sync implicitly before reading, scoped to its flags (`--source X` → that type; `--id`s → their prefixes; bare query → all types; `--input`-only → none), quiet unless something was ingested, degrading to the cache as-is if sync fails; `--no-sync` opts out. `p import` and `share` record what they write: every session derive carries a provenance `ArtifactRef` (stamped *before* the source is read, in `DerivedDoc.provenance`), and the cache-write sites call `sync::record_artifact` so the next sync sees those artifacts as unchanged instead of re-deriving them. Every import flow — explicit `--session`, picker multi-select, `--all`, and the most-recent fallbacks — loops the per-session helpers, so every session write is recorded; there is no bulk `derive_project` path in the CLI anymore, and `p import pi --all` now emits one Path per session like every other provider (it used to emit a single combined Graph). `--no-cache` paths record nothing: the manifest describes the cache. - `ArtifactType` (`crates/path-cli/src/artifact.rs`) is the general enum naming artifact sources — the seven agent harnesses (incl. copilot) plus `Git` (8 variants). Git artifacts are *recorded* in the manifest by `p import git` (id `-`, `path` = the repo directory) but never *discovered* — there is no machine-wide registry of repos — so sync reports them and leaves them alone. Github and pathbase are deliberately not artifact types: they are remote services, not local artifact sources, and their imports stay out of the manifest. It derives `clap::ValueEnum` and is used by `p cache sync` types, the sync manifest keys, `ArtifactRow.artifact_type`, and `cmd_import`'s cache-id prefixes (`name()` is both the manifest key and the `make_id` source string). The deliberately parallel `Harness` enum (`crates/path-cli/src/harness.rs`, alongside `HarnessBundle`) names the seven agent *runtimes* — things sessions can be shared from and resumed into — and is what `share`/`resume` `--harness` take, so future non-harness artifact types stay unrepresentable there (you can't resume into a git repo). `Harness::artifact_type()` maps into the general enum; `ArtifactType::harness()` is the partial inverse. Keep new code on `ArtifactType` unless it's genuinely harness-only. diff --git a/Cargo.lock b/Cargo.lock index 3305f9a8..f769fec1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2466,7 +2466,7 @@ dependencies = [ [[package]] name = "path-cli" -version = "0.17.0" +version = "0.18.0" dependencies = [ "anyhow", "assert_cmd", diff --git a/Cargo.toml b/Cargo.toml index d7103cdd..1e65edb7 100644 --- a/Cargo.toml +++ b/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.17.0", path = "crates/path-cli" } +path-cli = { version = "0.18.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"] } diff --git a/crates/path-cli/Cargo.toml b/crates/path-cli/Cargo.toml index b42638a5..81d425ac 100644 --- a/crates/path-cli/Cargo.toml +++ b/crates/path-cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "path-cli" -version = "0.17.0" +version = "0.18.0" edition.workspace = true license.workspace = true repository = "https://github.com/empathic/toolpath" diff --git a/crates/path-cli/src/cache.rs b/crates/path-cli/src/cache.rs index 8203f67e..5503cd72 100644 --- a/crates/path-cli/src/cache.rs +++ b/crates/path-cli/src/cache.rs @@ -83,6 +83,13 @@ pub(crate) fn write_cached(id: &str, doc: &Graph, force: bool) -> Result Result<()> { if let Err(e) = crate::sync::evict_cache_id(id) { eprintln!("warning: sync manifest not updated: {e}"); } + #[cfg(not(target_os = "emscripten"))] + if let Err(e) = crate::query::index::purge_id(id) { + eprintln!("warning: query index not updated: {e}"); + } eprintln!("Removed {id}"); Ok(()) } diff --git a/crates/path-cli/src/query/index.rs b/crates/path-cli/src/query/index.rs new file mode 100644 index 00000000..480ac68d --- /dev/null +++ b/crates/path-cli/src/query/index.rs @@ -0,0 +1,481 @@ +//! SQLite step index over the cache files at `$CONFIG_DIR/index.db`. +//! +//! The cache files stay canonical; this database is a disposable +//! accelerator holding every wrapped step as a row (`json` — exactly what +//! `wrap_graph` emits, compact) plus generated columns for the hot fields +//! the pushdown predicates filter on. Delete the file and the next query +//! rebuilds it lazily; bump [`SCHEMA_VERSION`] and it rebuilds itself. +//! +//! Freshness is stat-level: each indexed doc records its file's +//! `(mtime_ns, size)`, and a query serves rows only when a fresh stat +//! matches — otherwise the doc is reparsed and reindexed in place. Every +//! `write_cached` also indexes what it just wrote ([`index_written_doc`]), +//! so sync/import keep the index warm. +//! +//! Connections are per-thread (rayon workers reindex concurrently; WAL + +//! busy timeout serialize the writers), keyed by database path so tests +//! that repoint `$TOOLPATH_CONFIG_DIR` get fresh connections. + +use anyhow::{Context, Result}; +use rusqlite::Connection; +use std::path::{Path, PathBuf}; + +use super::plan::RowPredicate; +use crate::config::config_dir; + +const INDEX_FILE: &str = "index.db"; +const SCHEMA_VERSION: i64 = 1; + +/// A handle to the index: just its resolved path plus the guarantee that +/// the schema exists (ensured once, on the thread that opened it). +/// Cloneable into worker closures; actual connections are per-thread. +#[derive(Debug, Clone)] +pub struct StepIndex { + db_path: PathBuf, +} + +/// Open (creating/migrating if needed) the default index. `None` when +/// disabled via `TOOLPATH_QUERY_NO_INDEX` or when the database cannot be +/// opened — the caller falls back to plain file parsing. +pub fn open_default() -> Option { + if matches!(std::env::var("TOOLPATH_QUERY_NO_INDEX"), Ok(v) if !v.is_empty() && v != "0") { + return None; + } + let db_path = config_dir().ok()?.join(INDEX_FILE); + match ensure_schema(&db_path) { + Ok(()) => Some(StepIndex { db_path }), + Err(e) => { + eprintln!("warning: query index unavailable: {e:#}"); + None + } + } +} + +impl StepIndex { + /// Rows for a doc whose recorded stamp matches `(mtime_ns, size)`, + /// parsed back to wrapped-step values (prefiltered by `pred` when + /// given). `None` means stale or never indexed — reparse the file. + pub fn fresh_steps( + &self, + cache_id: &str, + mtime_ns: i64, + size: u64, + pred: Option<&RowPredicate>, + ) -> Result>> { + self.with_conn(|conn| { + if !stamp_matches(conn, cache_id, mtime_ns, size)? { + return Ok(None); + } + let (where_sql, params) = predicate_sql(pred); + let sql = format!("SELECT json FROM steps WHERE cache_id = ?1{where_sql} ORDER BY seq"); + let mut stmt = conn.prepare_cached(&sql)?; + let mut bound: Vec<&dyn rusqlite::ToSql> = vec![&cache_id]; + for p in ¶ms { + bound.push(p); + } + let mut rows = stmt.query(&bound[..])?; + let mut steps = Vec::new(); + while let Some(row) = rows.next()? { + let json: String = row.get(0)?; + steps.push(serde_json::from_str(&json).context("parse indexed step")?); + } + Ok(Some(steps)) + }) + } + + /// `SELECT count(*)` for a fresh doc — the fully absorbed path: no row + /// bytes are read or parsed. `None` means stale or never indexed. + pub fn fresh_count( + &self, + cache_id: &str, + mtime_ns: i64, + size: u64, + pred: Option<&RowPredicate>, + ) -> Result> { + self.with_conn(|conn| { + if !stamp_matches(conn, cache_id, mtime_ns, size)? { + return Ok(None); + } + let (where_sql, params) = predicate_sql(pred); + let sql = format!("SELECT count(*) FROM steps WHERE cache_id = ?1{where_sql}"); + let mut stmt = conn.prepare_cached(&sql)?; + let mut bound: Vec<&dyn rusqlite::ToSql> = vec![&cache_id]; + for p in ¶ms { + bound.push(p); + } + let n: i64 = stmt.query_row(&bound[..], |row| row.get(0))?; + Ok(Some(n)) + }) + } + + /// Replace a doc's rows and stamp in one transaction. + pub fn store( + &self, + cache_id: &str, + steps: &[serde_json::Value], + mtime_ns: i64, + size: u64, + ) -> Result<()> { + self.with_conn(|conn| { + let tx = conn.unchecked_transaction()?; + tx.execute("DELETE FROM steps WHERE cache_id = ?1", [cache_id])?; + tx.execute( + "INSERT OR REPLACE INTO documents (cache_id, mtime_ns, size) VALUES (?1, ?2, ?3)", + rusqlite::params![cache_id, mtime_ns, size as i64], + )?; + { + let mut ins = tx.prepare_cached( + "INSERT INTO steps (cache_id, seq, json) VALUES (?1, ?2, ?3)", + )?; + for (seq, step) in steps.iter().enumerate() { + ins.execute(rusqlite::params![ + cache_id, + seq as i64, + serde_json::to_string(step)? + ])?; + } + } + tx.commit()?; + Ok(()) + }) + } + + /// Drop a doc's rows and stamp (evicted or unreadable doc). + pub fn purge(&self, cache_id: &str) -> Result<()> { + self.with_conn(|conn| { + let tx = conn.unchecked_transaction()?; + tx.execute("DELETE FROM steps WHERE cache_id = ?1", [cache_id])?; + tx.execute("DELETE FROM documents WHERE cache_id = ?1", [cache_id])?; + tx.commit()?; + Ok(()) + }) + } + + /// Drop rows for docs no longer in the cache listing. Called on full + /// scans, where `keep` really is the complete cache. + pub fn retain(&self, keep: &std::collections::HashSet<&str>) -> Result<()> { + self.with_conn(|conn| { + let mut stmt = conn.prepare("SELECT cache_id FROM documents")?; + let known: Vec = stmt + .query_map([], |row| row.get(0))? + .collect::>()?; + drop(stmt); + let tx = conn.unchecked_transaction()?; + for id in known.iter().filter(|id| !keep.contains(id.as_str())) { + tx.execute("DELETE FROM steps WHERE cache_id = ?1", [id])?; + tx.execute("DELETE FROM documents WHERE cache_id = ?1", [id])?; + } + tx.commit()?; + Ok(()) + }) + } + + /// Run `f` with this thread's connection to `self.db_path`, opening + /// one if the thread has none (or has one for a different path). The + /// connection is taken out of the slot while `f` runs, so a nested call + /// opens a short-lived second connection instead of panicking on the + /// `RefCell`. + fn with_conn(&self, f: impl FnOnce(&Connection) -> Result) -> Result { + use std::cell::RefCell; + thread_local! { + static CONN: RefCell> = const { RefCell::new(None) }; + } + CONN.with(|cell| { + let conn = match cell.borrow_mut().take() { + Some((p, c)) if p == self.db_path => c, + _ => open_conn(&self.db_path)?, + }; + let result = f(&conn); + *cell.borrow_mut() = Some((self.db_path.clone(), conn)); + result + }) + } +} + +/// The file's identity stamp: `(mtime_ns, size)`. Any stat failure reads +/// as "unknowable", which never matches a recorded stamp. +pub fn file_stamp(path: &Path) -> Option<(i64, u64)> { + let meta = std::fs::metadata(path).ok()?; + let mtime = meta.modified().ok()?; + let ns = mtime + .duration_since(std::time::UNIX_EPOCH) + .ok()? + .as_nanos() + .try_into() + .ok()?; + Some((ns, meta.len())) +} + +/// Write-through for `cache::write_cached`: index the doc that was just +/// written, stamped with the written file's own stat. +pub fn index_written_doc(cache_id: &str, doc: &toolpath::v1::Graph, path: &Path) -> Result<()> { + let Some(index) = open_default() else { + return Ok(()); + }; + let Some((mtime_ns, size)) = file_stamp(path) else { + return Ok(()); + }; + let steps = super::wrap_document(cache_id, doc); + index.store(cache_id, &steps, mtime_ns, size) +} + +/// Purge hook for `p cache rm`. +pub fn purge_id(cache_id: &str) -> Result<()> { + if let Some(index) = open_default() { + index.purge(cache_id)?; + } + Ok(()) +} + +fn stamp_matches(conn: &Connection, cache_id: &str, mtime_ns: i64, size: u64) -> Result { + let mut stmt = + conn.prepare_cached("SELECT mtime_ns, size FROM documents WHERE cache_id = ?1")?; + let stamp: Option<(i64, i64)> = stmt + .query_row([cache_id], |row| Ok((row.get(0)?, row.get(1)?))) + .map(Some) + .or_else(|e| match e { + rusqlite::Error::QueryReturnedNoRows => Ok(None), + e => Err(e), + })?; + Ok(stamp == Some((mtime_ns, size as i64))) +} + +/// A predicate's SQL over the generated columns, as (`" AND …"`, params). +/// Parameters continue from `?2` (`?1` is always `cache_id`). Each atom is +/// exactly its jq counterpart on wrapper-produced rows — see +/// [`RowPredicate`]. +fn predicate_sql(pred: Option<&RowPredicate>) -> (String, Vec) { + let mut params = Vec::new(); + let sql = match pred { + None => String::new(), + Some(p) => format!(" AND {}", predicate_clause(p, &mut params)), + }; + (sql, params) +} + +fn predicate_clause(pred: &RowPredicate, params: &mut Vec) -> String { + match pred { + RowPredicate::DeadEnd(b) => format!("dead_end = {}", i32::from(*b)), + RowPredicate::ActorEq(s) => { + params.push(s.clone()); + format!("actor = ?{}", params.len() + 1) + } + RowPredicate::ActorStartsWith(s) => { + // `substr` counts characters, so this is exact byte-prefix + // equality for any valid UTF-8 prefix (unlike LIKE, which is + // ASCII-case-insensitive, or GLOB, which interprets * ? [). + params.push(s.clone()); + format!( + "substr(actor, 1, {}) = ?{}", + s.chars().count(), + params.len() + 1 + ) + } + RowPredicate::SourceEq(s) => { + params.push(s.clone()); + format!("source = ?{}", params.len() + 1) + } + RowPredicate::And(l, r) => { + let l = predicate_clause(l, params); + let r = predicate_clause(r, params); + format!("({l} AND {r})") + } + } +} + +fn open_conn(db_path: &Path) -> Result { + let conn = Connection::open(db_path).with_context(|| format!("open {}", db_path.display()))?; + conn.busy_timeout(std::time::Duration::from_secs(5))?; + conn.pragma_update(None, "journal_mode", "WAL")?; + Ok(conn) +} + +/// Create/refresh the schema. A version mismatch drops everything — the +/// index is derived state, so a rebuild is the migration. +fn ensure_schema(db_path: &Path) -> Result<()> { + if let Some(dir) = db_path.parent() { + std::fs::create_dir_all(dir).with_context(|| format!("create {}", dir.display()))?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let _ = std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700)); + } + } + let conn = open_conn(db_path)?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let _ = std::fs::set_permissions(db_path, std::fs::Permissions::from_mode(0o600)); + } + let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?; + if version != SCHEMA_VERSION { + conn.execute_batch("DROP TABLE IF EXISTS steps; DROP TABLE IF EXISTS documents;")?; + } + conn.execute_batch(&format!( + "CREATE TABLE IF NOT EXISTS documents ( + cache_id TEXT PRIMARY KEY, + mtime_ns INTEGER NOT NULL, + size INTEGER NOT NULL + ); + CREATE TABLE IF NOT EXISTS steps ( + cache_id TEXT NOT NULL, + seq INTEGER NOT NULL, + json TEXT NOT NULL, + dead_end INT GENERATED ALWAYS AS (json_extract(json, '$.dead_end')) VIRTUAL, + actor TEXT GENERATED ALWAYS AS (json_extract(json, '$.step.actor')) VIRTUAL, + source TEXT GENERATED ALWAYS AS (json_extract(json, '$.path.meta.source')) VIRTUAL, + PRIMARY KEY (cache_id, seq) + ); + CREATE INDEX IF NOT EXISTS steps_dead_end ON steps (cache_id, dead_end); + CREATE INDEX IF NOT EXISTS steps_actor ON steps (cache_id, actor); + CREATE INDEX IF NOT EXISTS steps_source ON steps (cache_id, source); + PRAGMA user_version = {SCHEMA_VERSION};" + ))?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::{CONFIG_DIR_ENV, TEST_ENV_LOCK}; + use serde_json::json; + + fn with_cfg R, R>(f: F) -> R { + let temp = tempfile::tempdir().unwrap(); + let _g = TEST_ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + unsafe { + std::env::set_var(CONFIG_DIR_ENV, temp.path()); + } + let result = f(temp.path()); + unsafe { + std::env::remove_var(CONFIG_DIR_ENV); + } + result + } + + fn rows() -> Vec { + vec![ + json!({"cache_id": "d", "step": {"id": "s1", "actor": "agent:claude"}, "dead_end": false, + "path": {"id": "p", "meta": {"source": "claude-code"}}}), + json!({"cache_id": "d", "step": {"id": "s2", "actor": "human:ben"}, "dead_end": true, + "path": {"id": "p", "meta": {"source": "claude-code"}}}), + json!({"cache_id": "d", "step": {"id": "s3", "actor": "agent:codex"}, "dead_end": true, + "path": {"id": "p", "meta": {"source": "codex"}}}), + ] + } + + #[test] + fn store_and_fresh_roundtrip() { + with_cfg(|_| { + let index = open_default().unwrap(); + index.store("d", &rows(), 42, 100).unwrap(); + + // Matching stamp serves the rows verbatim, in order. + let served = index.fresh_steps("d", 42, 100, None).unwrap().unwrap(); + assert_eq!(served, rows()); + + // Any stamp drift reads as stale. + assert!(index.fresh_steps("d", 43, 100, None).unwrap().is_none()); + assert!(index.fresh_steps("d", 42, 101, None).unwrap().is_none()); + assert!(index.fresh_steps("other", 42, 100, None).unwrap().is_none()); + }); + } + + #[test] + fn predicates_filter_rows_and_counts() { + with_cfg(|_| { + let index = open_default().unwrap(); + index.store("d", &rows(), 1, 1).unwrap(); + + let dead = RowPredicate::DeadEnd(true); + let served = index.fresh_steps("d", 1, 1, Some(&dead)).unwrap().unwrap(); + assert_eq!(served.len(), 2); + assert_eq!(index.fresh_count("d", 1, 1, Some(&dead)).unwrap(), Some(2)); + + let agent = RowPredicate::ActorStartsWith("agent:".into()); + assert_eq!(index.fresh_count("d", 1, 1, Some(&agent)).unwrap(), Some(2)); + + let both = RowPredicate::And(Box::new(dead), Box::new(agent)); + let served = index.fresh_steps("d", 1, 1, Some(&both)).unwrap().unwrap(); + assert_eq!(served.len(), 1); + assert_eq!(served[0]["step"]["id"], "s3"); + + let src = RowPredicate::SourceEq("claude-code".into()); + assert_eq!(index.fresh_count("d", 1, 1, Some(&src)).unwrap(), Some(2)); + + let eq = RowPredicate::ActorEq("human:ben".into()); + assert_eq!(index.fresh_count("d", 1, 1, Some(&eq)).unwrap(), Some(1)); + + assert_eq!(index.fresh_count("d", 1, 1, None).unwrap(), Some(3)); + }); + } + + #[test] + fn store_replaces_and_purge_forgets() { + with_cfg(|_| { + let index = open_default().unwrap(); + index.store("d", &rows(), 1, 1).unwrap(); + index.store("d", &rows()[..1], 2, 2).unwrap(); + let served = index.fresh_steps("d", 2, 2, None).unwrap().unwrap(); + assert_eq!(served.len(), 1, "store replaces prior rows"); + + index.purge("d").unwrap(); + assert!(index.fresh_steps("d", 2, 2, None).unwrap().is_none()); + }); + } + + #[test] + fn retain_drops_absent_docs() { + with_cfg(|_| { + let index = open_default().unwrap(); + index.store("keep", &rows(), 1, 1).unwrap(); + index.store("gone", &rows(), 1, 1).unwrap(); + let keep: std::collections::HashSet<&str> = ["keep"].into(); + index.retain(&keep).unwrap(); + assert!(index.fresh_steps("keep", 1, 1, None).unwrap().is_some()); + assert!(index.fresh_steps("gone", 1, 1, None).unwrap().is_none()); + }); + } + + #[test] + fn empty_doc_is_fresh_with_zero_rows() { + with_cfg(|_| { + let index = open_default().unwrap(); + index.store("empty", &[], 1, 1).unwrap(); + let served = index.fresh_steps("empty", 1, 1, None).unwrap().unwrap(); + assert!(served.is_empty(), "fresh but empty ≠ stale"); + assert_eq!(index.fresh_count("empty", 1, 1, None).unwrap(), Some(0)); + }); + } + + #[test] + fn schema_version_bump_rebuilds() { + with_cfg(|_| { + let index = open_default().unwrap(); + index.store("d", &rows(), 1, 1).unwrap(); + index + .with_conn(|conn| { + conn.pragma_update(None, "user_version", 999)?; + Ok(()) + }) + .unwrap(); + // Reopening sees the foreign version and starts over. + let index = open_default().unwrap(); + assert!(index.fresh_steps("d", 1, 1, None).unwrap().is_none()); + }); + } + + #[test] + fn no_index_env_disables() { + with_cfg(|_| { + unsafe { + std::env::set_var("TOOLPATH_QUERY_NO_INDEX", "1"); + } + let disabled = open_default(); + unsafe { + std::env::remove_var("TOOLPATH_QUERY_NO_INDEX"); + } + assert!(disabled.is_none()); + }); + } +} diff --git a/crates/path-cli/src/query/mod.rs b/crates/path-cli/src/query/mod.rs index f833de1e..6542aee5 100644 --- a/crates/path-cli/src/query/mod.rs +++ b/crates/path-cli/src/query/mod.rs @@ -8,6 +8,8 @@ //! documents load and hands jaq the array. mod filter; +#[cfg(not(target_os = "emscripten"))] +pub(crate) mod index; mod plan; use anyhow::{Context, Result}; @@ -51,20 +53,160 @@ pub struct Scope { /// path, which is still lean — the step values are held once, not re-serialized. pub fn run(scope: &Scope, code: &str, compact: bool, raw: bool) -> Result<()> { let plan = plan::analyze(code); + let ctx = LoadCtx::new(scope, code); // Opt-in observability: `TOOLPATH_QUERY_EXPLAIN=1` reveals the execution // strategy on stderr. Not a behavioral flag — purely diagnostic. let explain = std::env::var("TOOLPATH_QUERY_EXPLAIN"); if matches!(explain.as_deref(), Ok(v) if !v.is_empty() && v != "0") { eprintln!("query plan: {}", plan.describe()); + #[cfg(not(target_os = "emscripten"))] + eprintln!("query index: {}", ctx.describe(scope, code)); } // Buffer stdout: the streaming path prints one value per output, and a // raw `StdoutLock` is line-buffered (a syscall per line). let stdout = std::io::stdout(); let mut out = std::io::BufWriter::new(stdout.lock()); - execute_plan(scope, &plan, code, compact, raw, &mut out)?; + execute_plan(scope, &ctx, &plan, code, compact, raw, &mut out)?; out.flush().context("flush stdout") } +/// Everything a per-document load needs: the content-scope filters, and (off +/// wasm) the step index plus the filter's recognized leading predicate. +/// Content-scoped queries (`--kind`/`--project`/`--parent-dir`) bypass the +/// index entirely — its rows are wrapped unfiltered, and the path-level +/// filters (with their canonicalization) run during wrapping — so those +/// take the parse path where the filters already live. +struct LoadCtx { + kind_sel: Option, + project: Option, + parent_dir: Option, + #[cfg(not(target_os = "emscripten"))] + index: Option, + #[cfg(not(target_os = "emscripten"))] + pred: Option, +} + +impl LoadCtx { + fn new(scope: &Scope, code: &str) -> Self { + #[cfg(target_os = "emscripten")] + let _ = code; + #[cfg(not(target_os = "emscripten"))] + let index = if scope.kind.is_some() || scope.project.is_some() || scope.parent_dir.is_some() + { + None + } else { + index::open_default() + }; + Self { + kind_sel: scope.kind.as_deref().map(kinds::parse_kind_selector), + project: scope.project.as_deref().map(canonicalize_or_self), + parent_dir: scope.parent_dir.as_deref().map(canonicalize_or_self), + #[cfg(not(target_os = "emscripten"))] + pred: index.as_ref().and_then(|_| plan::row_predicate(code)), + #[cfg(not(target_os = "emscripten"))] + index, + } + } + + /// One line for `TOOLPATH_QUERY_EXPLAIN`. + #[cfg(not(target_os = "emscripten"))] + fn describe(&self, scope: &Scope, code: &str) -> String { + let Some(_) = &self.index else { + return "off (disabled, unavailable, or content-scoped)".to_string(); + }; + if scope.inputs.is_empty() + && let Some(pred) = plan::absorbable_count(code) + { + return match pred { + Some(p) => format!("count absorbed into SQL ({})", p.describe()), + None => "count absorbed into SQL".to_string(), + }; + } + match &self.pred { + Some(p) => format!("serving fresh docs, prefilter {}", p.describe()), + None => "serving fresh docs".to_string(), + } + } +} + +/// One document's wrapped steps: served from the index when the doc is +/// fresh (prefiltered in SQL), else parsed from the file — reindexing it in +/// passing when it was merely stale. Never touches the index for `--input` +/// files or stdin. +fn doc_steps(ctx: &LoadCtx, src: &DocSource) -> Result> { + #[cfg(not(target_os = "emscripten"))] + if src.indexed + && let Some(step_index) = &ctx.index + && let SourceLoc::File(path) = &src.location + && let Some((mtime_ns, size)) = index::file_stamp(path) + { + if let Some(steps) = + step_index.fresh_steps(&src.cache_id, mtime_ns, size, ctx.pred.as_ref())? + { + return Ok(steps); + } + // Stale (or never indexed): the stamp was taken before the read, so + // a write racing the parse re-serves from the file next query + // instead of going unnoticed. + let steps = load_and_wrap(src, None, None, None)?; + if let Err(e) = step_index.store(&src.cache_id, &steps, mtime_ns, size) { + eprintln!( + "warning: query index not updated for {}: {e:#}", + src.cache_id + ); + } + return Ok(match &ctx.pred { + Some(p) => steps.into_iter().filter(|s| p.matches(s)).collect(), + None => steps, + }); + } + load_and_wrap( + src, + ctx.kind_sel.as_ref(), + ctx.project.as_deref(), + ctx.parent_dir.as_deref(), + ) +} + +/// One document's step count under the absorbed-count path: `SELECT +/// count(*)` for a fresh doc (no row bytes read), else the length of +/// `doc_steps` (which reindexes and prefilters with the same predicate). +#[cfg(not(target_os = "emscripten"))] +fn doc_count(ctx: &LoadCtx, src: &DocSource) -> Result { + if src.indexed + && let Some(step_index) = &ctx.index + && let SourceLoc::File(path) = &src.location + && let Some((mtime_ns, size)) = index::file_stamp(path) + && let Some(n) = step_index.fresh_count(&src.cache_id, mtime_ns, size, ctx.pred.as_ref())? + { + return Ok(n); + } + Ok(doc_steps(ctx, src)?.len() as i64) +} + +/// Wrap a whole document's steps with no content scoping — what the index +/// stores. Used by the `write_cached` write-through. +#[cfg(not(target_os = "emscripten"))] +pub(crate) fn wrap_document(cache_id: &str, graph: &Graph) -> Vec { + let mut steps = Vec::new(); + wrap_graph(cache_id, graph, None, None, None, &mut steps); + steps +} + +/// On a full-cache scan, drop index rows for docs no longer in the cache +/// (`p cache rm` purges eagerly; this catches out-of-band deletions). Only a +/// full scan sees the complete listing, so only it may prune. +#[cfg(not(target_os = "emscripten"))] +fn maybe_retain(scope: &Scope, ctx: &LoadCtx, sources: &[DocSource]) { + let full_scan = scope.source.is_none() && scope.ids.is_empty() && scope.inputs.is_empty(); + if full_scan && let Some(step_index) = &ctx.index { + let keep: HashSet<&str> = sources.iter().map(|s| s.cache_id.as_str()).collect(); + if let Err(e) = step_index.retain(&keep) { + eprintln!("warning: query index not pruned: {e:#}"); + } + } +} + /// Dispatch a plan to its driver. `PerFileStream` and `Decompose` treat every /// file independently, so their whole per-file pipeline — parse, wrap, filter, /// render/pack — runs on worker threads and only ordered consumption stays @@ -73,21 +215,43 @@ pub fn run(scope: &Scope, code: &str, compact: bool, raw: bool) -> Result<()> { #[cfg(not(target_os = "emscripten"))] fn execute_plan( scope: &Scope, + ctx: &LoadCtx, plan: &plan::Plan, code: &str, compact: bool, raw: bool, out: &mut dyn Write, ) -> Result<()> { + // Fully absorbed count: the whole filter is a (possibly predicated) + // `length`, so fresh docs answer with `SELECT count(*)` and no step is + // ever parsed. `--input` docs would still need jaq, so their presence + // falls through to the general drivers. + if ctx.index.is_some() && scope.inputs.is_empty() && plan::absorbable_count(code).is_some() { + filter::compile_check(code)?; + let mut total: i64 = 0; + for_each_file( + scope, + ctx, + |src| doc_count(ctx, src), + |n| { + total += n; + Ok(()) + }, + )?; + writeln!(out, "{total}")?; + return Ok(()); + } + match plan { plan::Plan::Slurp => filter::execute(plan, code, compact, raw, out, |emit| { - stream_files(scope, emit) + stream_files(scope, ctx, emit) }), plan::Plan::PerFileStream => { filter::compile_check(code)?; for_each_file( scope, - |steps| filter::render_file(code, steps, compact, raw), + ctx, + |src| filter::render_file(code, doc_steps(ctx, src)?, compact, raw), |bytes| { out.write_all(&bytes)?; Ok(()) @@ -100,7 +264,8 @@ fn execute_plan( let mut saw_file = false; for_each_file( scope, - |steps| filter::partials_file(code, steps), + ctx, + |src| filter::partials_file(code, doc_steps(ctx, src)?), |bytes| { saw_file = true; partials.extend(filter::unpack_partials(&bytes)?); @@ -117,6 +282,7 @@ fn execute_plan( #[cfg(target_os = "emscripten")] fn execute_plan( scope: &Scope, + ctx: &LoadCtx, plan: &plan::Plan, code: &str, compact: bool, @@ -124,7 +290,7 @@ fn execute_plan( out: &mut dyn Write, ) -> Result<()> { filter::execute(plan, code, compact, raw, out, |emit| { - stream_files(scope, emit) + stream_files(scope, ctx, emit) }) } @@ -138,30 +304,18 @@ fn execute_plan( #[cfg(not(target_os = "emscripten"))] fn for_each_file( scope: &Scope, - per_file: impl Fn(Vec) -> Result + Sync, + ctx: &LoadCtx, + per_file: impl Fn(&DocSource) -> Result + Sync, mut consume: impl FnMut(T) -> Result<()>, ) -> Result<()> { use rayon::prelude::*; - let kind_sel = scope.kind.as_deref().map(kinds::parse_kind_selector); - let project = scope.project.as_deref().map(canonicalize_or_self); - let parent_dir = scope.parent_dir.as_deref().map(canonicalize_or_self); let sources = select_files(scope)?; + maybe_retain(scope, ctx, &sources); let chunk = rayon::current_num_threads().max(1) * 2; for batch in sources.chunks(chunk) { - let products: Vec> = batch - .par_iter() - .map(|src| { - load_and_wrap( - src, - kind_sel.as_ref(), - project.as_deref(), - parent_dir.as_deref(), - ) - .and_then(&per_file) - }) - .collect(); + let products: Vec> = batch.par_iter().map(&per_file).collect(); for (src, res) in batch.iter().zip(products) { match res { Ok(product) => consume(product)?, @@ -183,6 +337,11 @@ struct DocSource { /// or parse failure is an error — not the skip-with-warning that's right /// for a corrupt file encountered during the whole-cache scan. explicit: bool, + /// A cache document — the only kind the step index may serve or store. + /// `--input` files and stdin are never indexed. (Read only off wasm; + /// the emscripten build has no index.) + #[cfg_attr(target_os = "emscripten", allow(dead_code))] + indexed: bool, } enum SourceLoc { @@ -205,28 +364,21 @@ impl DocSource { /// thread because jaq values are `Rc`-based, not `Send`. Chunking keeps /// output (and per-file warnings) byte-identical to a sequential scan while /// holding at most one chunk of parsed documents in memory. -fn stream_files(scope: &Scope, emit: &mut dyn FnMut(Val) -> Result<()>) -> Result<()> { - let kind_sel = scope.kind.as_deref().map(kinds::parse_kind_selector); - let project = scope.project.as_deref().map(canonicalize_or_self); - let parent_dir = scope.parent_dir.as_deref().map(canonicalize_or_self); +fn stream_files( + scope: &Scope, + ctx: &LoadCtx, + emit: &mut dyn FnMut(Val) -> Result<()>, +) -> Result<()> { let sources = select_files(scope)?; #[cfg(not(target_os = "emscripten"))] { use rayon::prelude::*; + maybe_retain(scope, ctx, &sources); let chunk = rayon::current_num_threads().max(1) * 2; for batch in sources.chunks(chunk) { - let parsed: Vec>> = batch - .par_iter() - .map(|src| { - load_and_wrap( - src, - kind_sel.as_ref(), - project.as_deref(), - parent_dir.as_deref(), - ) - }) - .collect(); + let parsed: Vec>> = + batch.par_iter().map(|src| doc_steps(ctx, src)).collect(); for (src, res) in batch.iter().zip(parsed) { emit_wrapped(src, res, emit)?; } @@ -236,12 +388,7 @@ fn stream_files(scope: &Scope, emit: &mut dyn FnMut(Val) -> Result<()>) -> Resul // The playground build has no threads (emscripten without pthreads). #[cfg(target_os = "emscripten")] for src in &sources { - let res = load_and_wrap( - src, - kind_sel.as_ref(), - project.as_deref(), - parent_dir.as_deref(), - ); + let res = doc_steps(ctx, src); emit_wrapped(src, res, emit)?; } @@ -257,7 +404,14 @@ fn load_and_wrap( ) -> Result> { let graph = read_source(src)?; let mut steps = Vec::new(); - wrap_graph(src, &graph, kind_sel, project, parent_dir, &mut steps); + wrap_graph( + &src.cache_id, + &graph, + kind_sel, + project, + parent_dir, + &mut steps, + ); Ok(steps) } @@ -323,6 +477,7 @@ fn select_files(scope: &Scope) -> Result> { cache_id: entry.id, location: SourceLoc::File(entry.path), explicit: by_id, + indexed: true, }); } // Every requested `--id` must have matched a cached document. @@ -341,6 +496,7 @@ fn select_files(scope: &Scope) -> Result> { cache_id: "stdin".to_string(), location: SourceLoc::Stdin, explicit: true, + indexed: false, }); } else { // The full path as given, so two inputs sharing a basename @@ -350,6 +506,7 @@ fn select_files(scope: &Scope) -> Result> { cache_id: inp.clone(), location: SourceLoc::File(PathBuf::from(inp)), explicit: true, + indexed: false, }); } } @@ -384,7 +541,7 @@ fn read_source(src: &DocSource) -> Result { /// Walk a graph's inline paths, apply content scoping, and wrap surviving /// steps into `out`. fn wrap_graph( - src: &DocSource, + cache_id: &str, graph: &Graph, kind_sel: Option<&KindSelector>, project: Option<&FsPath>, @@ -413,12 +570,12 @@ fn wrap_graph( continue; } - wrap_path(src, path, out); + wrap_path(cache_id, path, out); } } /// Wrap every step of one path, computing the dead-end set once. -fn wrap_path(src: &DocSource, path: &Path, out: &mut Vec) { +fn wrap_path(cache_id: &str, path: &Path, out: &mut Vec) { let dead: HashSet<&str> = query::dead_ends(&path.steps, &path.path.head) .into_iter() .map(|s| s.step.id.as_str()) @@ -435,7 +592,7 @@ fn wrap_path(src: &DocSource, path: &Path, out: &mut Vec) { }; obj.insert( "cache_id".to_string(), - serde_json::Value::String(src.cache_id.clone()), + serde_json::Value::String(cache_id.to_string()), ); obj.insert("path".to_string(), path_ctx.clone()); obj.insert( @@ -492,14 +649,6 @@ mod tests { use super::*; use toolpath::v1::{Base, Graph, Path, PathIdentity, PathMeta, Step}; - fn doc_src(id: &str) -> DocSource { - DocSource { - cache_id: id.to_string(), - location: SourceLoc::Stdin, - explicit: false, - } - } - /// A path with a fork: s1 → {s2 → s3 (head), s2a (dead end)}. fn forked_path() -> Path { let s1 = Step::new("s1", "human:alex", "2026-01-01T10:00:00Z") @@ -533,7 +682,7 @@ mod tests { fn wraps_step_verbatim_with_context() { let path = forked_path(); let mut out = Vec::new(); - wrap_path(&doc_src("claude-abc"), &path, &mut out); + wrap_path("claude-abc", &path, &mut out); assert_eq!(out.len(), 4); let first = &out[0]; @@ -551,7 +700,7 @@ mod tests { fn dead_end_flag_tracks_ancestry_of_head() { let path = forked_path(); let mut out = Vec::new(); - wrap_path(&doc_src("g"), &path, &mut out); + wrap_path("g", &path, &mut out); let dead: std::collections::HashMap<&str, bool> = out .iter() @@ -573,12 +722,12 @@ mod tests { let graph = Graph::from_path(forked_path()); let mut out = Vec::new(); let sel = kinds::parse_kind_selector("agent-coding-session/v1.1.0"); - wrap_graph(&doc_src("g"), &graph, Some(&sel), None, None, &mut out); + wrap_graph("g", &graph, Some(&sel), None, None, &mut out); assert_eq!(out.len(), 4, "matching kind keeps all steps"); out.clear(); let miss = kinds::parse_kind_selector("agent-coding-session/v2"); - wrap_graph(&doc_src("g"), &graph, Some(&miss), None, None, &mut out); + wrap_graph("g", &graph, Some(&miss), None, None, &mut out); assert!(out.is_empty(), "non-matching kind drops the whole path"); } diff --git a/crates/path-cli/src/query/plan.rs b/crates/path-cli/src/query/plan.rs index 924eff48..69a8e85f 100644 --- a/crates/path-cli/src/query/plan.rs +++ b/crates/path-cli/src/query/plan.rs @@ -23,6 +23,8 @@ use jaq_core::load::{self, parse::BinaryOp, parse::Term}; use jaq_core::path::Part; +#[cfg(not(target_os = "emscripten"))] +use jaq_core::{load::lex::StrPart, ops::Cmp, path::Opt}; /// How to execute a filter over the cache. #[derive(Debug, Clone, PartialEq, Eq)] @@ -56,6 +58,213 @@ impl Plan { } } +/// A per-row predicate recognized from a filter's leading `select`, mappable +/// to SQL over the step index's generated columns. Recognition is exact, not +/// merely implied: on wrapper-produced rows (`dead_end` always a boolean, +/// `.step.actor` always a string) each atom means precisely its jq +/// counterpart, so one recognizer serves both index prefiltering and the +/// fully absorbed `count(*)` path. Anything not on this whitelist simply +/// isn't recognized — the query still runs, unprefiltered. +#[cfg(not(target_os = "emscripten"))] +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RowPredicate { + /// `select(.dead_end)` / `select(.dead_end | not)` + DeadEnd(bool), + /// `select(.step.actor == "…")` + ActorEq(String), + /// `select(.step.actor | startswith("…"))` + ActorStartsWith(String), + /// `select(.path.meta.source == "…")` + SourceEq(String), + /// `select(a and b)` + And(Box, Box), +} + +#[cfg(not(target_os = "emscripten"))] +impl RowPredicate { + /// The in-code mirror of the SQL clause, applied to rows that were just + /// reparsed from a stale file (the index's WHERE couldn't run). Must + /// agree with `index::predicate_clause` on every wrapper-produced row. + pub fn matches(&self, row: &serde_json::Value) -> bool { + match self { + RowPredicate::DeadEnd(b) => row["dead_end"].as_bool() == Some(*b), + RowPredicate::ActorEq(s) => row["step"]["actor"].as_str() == Some(s), + RowPredicate::ActorStartsWith(s) => row["step"]["actor"] + .as_str() + .is_some_and(|a| a.starts_with(s)), + RowPredicate::SourceEq(s) => row["path"]["meta"]["source"].as_str() == Some(s), + RowPredicate::And(l, r) => l.matches(row) && r.matches(row), + } + } + + /// Short form for `TOOLPATH_QUERY_EXPLAIN`. + pub fn describe(&self) -> String { + match self { + RowPredicate::DeadEnd(b) => format!("dead_end={b}"), + RowPredicate::ActorEq(s) => format!("actor={s:?}"), + RowPredicate::ActorStartsWith(s) => format!("actor^={s:?}"), + RowPredicate::SourceEq(s) => format!("source={s:?}"), + RowPredicate::And(l, r) => format!("{} and {}", l.describe(), r.describe()), + } + } +} + +/// The predicate established by the filter's first pipeline stage, if fully +/// recognized: `map(select(P)) | …`, `[.[] | select(P)] | …`, or +/// `.[] | select(P) | …`. Rows failing `P` are dropped by that first stage +/// before anything downstream can observe them, so prefiltering the input +/// with `P` leaves every later stage's input — and the final answer — +/// unchanged, whatever the plan. +#[cfg(not(target_os = "emscripten"))] +pub fn row_predicate(code: &str) -> Option { + let term = load::parse(code, |p| p.term())?; + let segs = pipe_segments(&term)?; + if let Some(body) = map_select_body(segs[0]) { + return predicate(body); + } + if is_iterate(segs[0]) + && let Some(second) = segs.get(1) + && let Some(body) = select_body(second) + { + return predicate(body); + } + None +} + +/// Whether the whole filter is a count the index can answer with +/// `SELECT count(*)`: bare `length`, or `map(select(P)) | length` (and the +/// collected-iteration spelling) with `P` fully recognized. Returns the +/// predicate to count under (`None` = count everything). +#[cfg(not(target_os = "emscripten"))] +pub fn absorbable_count(code: &str) -> Option> { + let term = load::parse(code, |p| p.term())?; + let segs = pipe_segments(&term)?; + let is_length = |t: &Term<&str>| matches!(t, Term::Call(name, args) if *name == "length" && args.is_empty()); + match segs.as_slice() { + [only] if is_length(only) => Some(None), + [head, tail] if is_length(tail) => { + let body = map_select_body(head)?; + Some(Some(predicate(body)?)) + } + _ => None, + } +} + +/// `select(P)` → `P`. +#[cfg(not(target_os = "emscripten"))] +fn select_body<'a, 's>(t: &'a Term<&'s str>) -> Option<&'a Term<&'s str>> { + match t { + Term::Call(name, args) if *name == "select" && args.len() == 1 => Some(&args[0]), + _ => None, + } +} + +/// `map(select(P))` or `[.[] | select(P)]` → `P`. +#[cfg(not(target_os = "emscripten"))] +fn map_select_body<'a, 's>(t: &'a Term<&'s str>) -> Option<&'a Term<&'s str>> { + if let Term::Call(name, args) = t + && *name == "map" + && args.len() == 1 + { + return select_body(&args[0]); + } + if let Term::Arr(Some(inner)) = t { + let segs = pipe_segments(inner)?; + if let [head, sel] = segs.as_slice() + && is_iterate(head) + { + return select_body(sel); + } + } + None +} + +#[cfg(not(target_os = "emscripten"))] +fn predicate(t: &Term<&str>) -> Option { + match t { + Term::BinOp(l, BinaryOp::And, r) => Some(RowPredicate::And( + Box::new(predicate(l)?), + Box::new(predicate(r)?), + )), + // `.step.actor == "…"` / `.path.meta.source == "…"` (either order). + Term::BinOp(l, BinaryOp::Cmp(Cmp::Eq), r) => { + let (path, lit) = match (field_path(l), str_lit(r)) { + (Some(p), Some(s)) => (p, s), + _ => (field_path(r)?, str_lit(l)?), + }; + match path.as_slice() { + ["step", "actor"] => Some(RowPredicate::ActorEq(lit)), + ["path", "meta", "source"] => Some(RowPredicate::SourceEq(lit)), + _ => None, + } + } + // `.step.actor | startswith("…")` + Term::BinOp(l, BinaryOp::Pipe(None), r) => match r.as_ref() { + Term::Call(name, args) + if *name == "startswith" + && args.len() == 1 + && field_path(l).as_deref() == Some(&["step", "actor"]) => + { + Some(RowPredicate::ActorStartsWith(str_lit(&args[0])?)) + } + Term::Call(name, args) + if *name == "not" + && args.is_empty() + && field_path(l).as_deref() == Some(&["dead_end"]) => + { + Some(RowPredicate::DeadEnd(false)) + } + _ => None, + }, + // `.dead_end` — truthiness; exact because the wrapper always writes + // a boolean. + _ => match field_path(t)?.as_slice() { + ["dead_end"] => Some(RowPredicate::DeadEnd(true)), + _ => None, + }, + } +} + +/// `.a.b.c` as `["a","b","c"]` — literal, non-optional field accesses only. +#[cfg(not(target_os = "emscripten"))] +fn field_path<'a>(t: &Term<&'a str>) -> Option> { + let Term::Path(inner, path) = t else { + return None; + }; + if !matches!(inner.as_ref(), Term::Id) { + return None; + } + let mut fields = Vec::with_capacity(path.0.len()); + for (part, opt) in &path.0 { + if !matches!(opt, Opt::Essential) { + return None; + } + let Part::Index(idx) = part else { + return None; + }; + fields.push(term_str(idx)?); + } + (!fields.is_empty()).then_some(fields) +} + +/// A plain string literal (no interpolation, no escapes, no format). +#[cfg(not(target_os = "emscripten"))] +fn term_str<'a>(t: &Term<&'a str>) -> Option<&'a str> { + match t { + Term::Str(None, parts) => match parts.as_slice() { + [StrPart::Str(s)] => Some(s), + [] => Some(""), + _ => None, + }, + _ => None, + } +} + +#[cfg(not(target_os = "emscripten"))] +fn str_lit(t: &Term<&str>) -> Option { + term_str(t).map(str::to_string) +} + /// Classify `code` into a [`Plan`]. Unparseable or unrecognized filters map to /// [`Plan::Slurp`] (the main executor re-parses and surfaces any real error). pub fn analyze(code: &str) -> Plan { @@ -325,4 +534,114 @@ mod tests { // `.` (whole array) has no decomposition; slurp (it's cheap anyway). assert_eq!(analyze("."), Plan::Slurp); } + + // ── Row-predicate recognition (index pushdown) ──────────────────── + + #[test] + fn recognizes_leading_select_predicates() { + assert_eq!( + row_predicate("map(select(.dead_end))"), + Some(RowPredicate::DeadEnd(true)) + ); + assert_eq!( + row_predicate(".[] | select(.dead_end | not) | .step.id"), + Some(RowPredicate::DeadEnd(false)) + ); + assert_eq!( + row_predicate("[.[] | select(.step.actor == \"human:ben\")] | length"), + Some(RowPredicate::ActorEq("human:ben".into())) + ); + assert_eq!( + row_predicate(r#"map(select(.step.actor | startswith("agent:")))"#), + Some(RowPredicate::ActorStartsWith("agent:".into())) + ); + assert_eq!( + row_predicate(r#"map(select(.path.meta.source == "claude-code")) | length"#), + Some(RowPredicate::SourceEq("claude-code".into())) + ); + assert_eq!( + row_predicate(r#"map(select(.dead_end and .step.actor == "agent:x"))"#), + Some(RowPredicate::And( + Box::new(RowPredicate::DeadEnd(true)), + Box::new(RowPredicate::ActorEq("agent:x".into())) + )) + ); + // Literal-first comparison order. + assert_eq!( + row_predicate(r#"map(select("agent:x" == .step.actor))"#), + Some(RowPredicate::ActorEq("agent:x".into())) + ); + } + + #[test] + fn unrecognized_predicates_stay_none() { + // A partially recognized conjunction is not used (whole-P only). + assert_eq!( + row_predicate("map(select(.dead_end and (.change | length > 2)))"), + None + ); + // Fields off the whitelist. + assert_eq!(row_predicate("map(select(.step.intent == \"x\"))"), None); + // Optional access, dynamic values, non-select heads. + assert_eq!(row_predicate("map(select(.dead_end?))"), None); + assert_eq!(row_predicate("map(select(.step.actor == .other))"), None); + assert_eq!(row_predicate("map(.step)"), None); + assert_eq!(row_predicate("group_by(.step.actor)"), None); + // select not in the first stage: prefiltering could change what the + // first stage sees, so it is not recognized. + assert_eq!(row_predicate("map(.step) | map(select(.dead_end))"), None); + // Interpolated strings are not literals. + assert_eq!( + row_predicate(r#"map(select(.step.actor == "a\(.x)"))"#), + None + ); + } + + #[test] + fn matches_mirrors_the_predicates() { + use serde_json::json; + let row = json!({ + "dead_end": true, + "step": {"actor": "agent:claude"}, + "path": {"meta": {"source": "claude-code"}} + }); + assert!(RowPredicate::DeadEnd(true).matches(&row)); + assert!(!RowPredicate::DeadEnd(false).matches(&row)); + assert!(RowPredicate::ActorEq("agent:claude".into()).matches(&row)); + assert!(RowPredicate::ActorStartsWith("agent:".into()).matches(&row)); + assert!(!RowPredicate::ActorStartsWith("human:".into()).matches(&row)); + assert!(RowPredicate::SourceEq("claude-code".into()).matches(&row)); + assert!( + RowPredicate::And( + Box::new(RowPredicate::DeadEnd(true)), + Box::new(RowPredicate::ActorStartsWith("agent:".into())) + ) + .matches(&row) + ); + // Missing source: `null == "x"` is false, like the SQL NULL. + let no_meta = json!({"dead_end": false, "step": {"actor": "a"}, "path": {"id": "p"}}); + assert!(!RowPredicate::SourceEq("claude-code".into()).matches(&no_meta)); + } + + #[test] + fn absorbable_counts() { + assert_eq!(absorbable_count("length"), Some(None)); + assert_eq!( + absorbable_count("map(select(.dead_end)) | length"), + Some(Some(RowPredicate::DeadEnd(true))) + ); + assert_eq!( + absorbable_count(r#"[.[] | select(.step.actor | startswith("agent:"))] | length"#), + Some(Some(RowPredicate::ActorStartsWith("agent:".into()))) + ); + // Not a bare count: extra stages, unrecognized predicates, other tails. + assert_eq!( + absorbable_count("map(select(.dead_end)) | length + 1"), + None + ); + assert_eq!(absorbable_count("map(select(.tokens > 3)) | length"), None); + assert_eq!(absorbable_count("map(.step) | length"), None); + assert_eq!(absorbable_count(".[] | length"), None); + assert_eq!(absorbable_count("length | tostring"), None); + } } diff --git a/crates/path-cli/tests/query.rs b/crates/path-cli/tests/query.rs index 80d83a12..43f68b81 100644 --- a/crates/path-cli/tests/query.rs +++ b/crates/path-cli/tests/query.rs @@ -623,3 +623,101 @@ fn query_parent_dir_scopes_sync_and_read() { .stdout(predicate::str::contains("0")) .stderr(predicate::str::contains("synced").not()); } + +// ── The SQLite step index ───────────────────────────────────────────── +// +// The index is a disposable accelerator: its output must be byte-identical +// to the plain file-parsing path, it must self-heal when a cache file +// changes behind it, and `TOOLPATH_QUERY_NO_INDEX` must bypass it. + +/// Stdout of a query run in `cfg`, optionally with the index disabled. +fn query_stdout(cfg: &Path, args: &[&str], no_index: bool) -> String { + let mut c = cmd(); + c.env("TOOLPATH_CONFIG_DIR", cfg); + if no_index { + c.env("TOOLPATH_QUERY_NO_INDEX", "1"); + } + let assert = c + .arg("query") + .arg("--no-sync") + .args(args) + .assert() + .success(); + String::from_utf8(assert.get_output().stdout.clone()).unwrap() +} + +#[test] +fn index_output_matches_file_path_for_every_plan() { + let cfg = sandbox(); + for args in [ + // Absorbed counts (bare and predicated). + &["length"] as &[&str], + &["map(select(.dead_end)) | length"], + // Decompose with a recognized prefilter, and without. + &["map(select(.step.actor | startswith(\"agent:\")))"], + &["map(.step.id)"], + // Stream with a recognized prefilter. + &[".[] | select(.dead_end) | .step.id"], + // Slurp (holistic) with a recognized prefilter in front. + &["map(select(.step.actor == \"human:alex\")) | group_by(.step.id) | length"], + // Slurp, nothing recognized. + &["group_by(.step.actor) | map({a: .[0].step.actor, n: length})"], + // Scoped to a source prefix. + &["--source", "claude", "map(.step.id)"], + // Raw string output. + &["--raw", ".[] | .step.actor"], + ] { + let plain = query_stdout(cfg.path(), args, true); + let first = query_stdout(cfg.path(), args, false); // builds the index + let second = query_stdout(cfg.path(), args, false); // serves from it + assert_eq!(first, plain, "first (indexing) run differs for {args:?}"); + assert_eq!(second, plain, "index-served run differs for {args:?}"); + } + assert!( + cfg.path().join("index.db").exists(), + "queries must have materialized the index" + ); +} + +#[test] +fn index_self_heals_when_a_cache_file_changes() { + let cfg = sandbox(); + assert_eq!(query_stdout(cfg.path(), &["length"], false).trim(), "6"); + + // The doc shrinks behind the index's back (git-pr42 has 2 steps). + std::fs::remove_file(cfg.path().join("documents/git-pr42.json")).unwrap(); + assert_eq!(query_stdout(cfg.path(), &["length"], false).trim(), "4"); + + // And a rewritten doc re-derives rather than serving stale rows. + seed(cfg.path(), "claude-sess1", GIT_DOC); + assert_eq!(query_stdout(cfg.path(), &["length"], false).trim(), "2"); +} + +#[test] +fn explain_reports_index_strategy() { + let cfg = sandbox(); + let assert = cmd() + .env("TOOLPATH_CONFIG_DIR", cfg.path()) + .env("TOOLPATH_QUERY_EXPLAIN", "1") + .args(["query", "--no-sync", "map(select(.dead_end)) | length"]) + .assert() + .success(); + let stderr = String::from_utf8(assert.get_output().stderr.clone()).unwrap(); + assert!( + stderr.contains("query index: count absorbed into SQL (dead_end=true)"), + "explain must name the absorbed strategy: {stderr}" + ); +} + +#[test] +fn cache_rm_purges_index_rows() { + let cfg = sandbox(); + query_stdout(cfg.path(), &["length"], false); + + cmd() + .env("TOOLPATH_CONFIG_DIR", cfg.path()) + .args(["p", "cache", "rm", "git-pr42"]) + .assert() + .success(); + assert_eq!(query_stdout(cfg.path(), &["length"], false).trim(), "4"); +} diff --git a/docs/design/graph-query-frontend.md b/docs/design/graph-query-frontend.md new file mode 100644 index 00000000..973b43a3 --- /dev/null +++ b/docs/design/graph-query-frontend.md @@ -0,0 +1,255 @@ +# A graph query frontend over local and Pathbase queries + +Status: design exploration, 2026-07. Nothing here is implemented. +Companion to `query-acceleration.md` (which explores making today's jaq +surface fast); this document explores a different axis — making local +and remote querying *the same thing*, using the query stack from +[facto](https://github.com/empathic/facto) (`graphdb`). + +## The gap + +Toolpath has two query models that share nothing: + +- **Local** (`path query`): jaq over a flat array of wrapped steps. + Good at step-content munging — projection, aggregation, selection + over the step JSON. Bad at anything relational: multi-hop DAG + traversal, delegation chains, joins across sessions. +- **Pathbase**: fixed-function surfaces. REST lists with `?limit` and + nothing else. Internal GraphQL exposes containment filtering + (`Path.steps(dataContains: …)` → JSONB `@>`) plus hand-carved + traversal resolvers — `Step.ancestors`, `Step.descendants`, + `Step.nearestAncestorWith(actorKind, changeType)`. Every new + relational question costs a new resolver and new SQL. + +The same question — "which sessions touched `src/main.rs` and what did +their dead-end branches try?" — is awkward-to-impossible in both, and +has to be asked in two unrelated languages depending on where the data +happens to live. + +`nearestAncestorWith` is the tell: traversal questions exist, and +today each one is answered by writing engine code. A graph query +frontend turns them into queries. + +## What facto provides + +facto (`graphdb`) is a TAO-style graph-serving engine with three query +languages — **GraphJQ** (jq-flavored pipelines), **Cypher** (openCypher +read subset), **SPARQL 1.1** — that all lower to one shared IR +(`graphdb-ir::query_ast::Query`) run by one planner + executor, with a +written denotational semantics (`GRAPHJQ-SEMANTICS.md`) that makes +planner rewrites provably answer-preserving. + +The properties that matter here: + +- **The engine is storage-agnostic.** `graphdb-engine::traversal:: + execute(reader, ext, config, graph_id, query)` — the + engine consumes only the `GraphReader` trait from `graphdb-core` + (~11 async read methods: `vertex_get`, `neighbor_ids_batch`, + `find_ids_by_predicate`, `get_object_doc`, …). SQL push-down + (`solution_query`) has a default returning `None`, so a host + implements only the core reads. Data never has to live in facto's + own Postgres. +- **The frontends are pure.** `graphdb-graphjq` / `graphdb-cypher` / + `graphdb-sparql` are parser crates (text → IR) that depend on + neither the engine nor any storage or async runtime. They compile + anywhere the CLI compiles, and the IR is plain serde JSON. +- **Bounded serving semantics.** Depth clamps, frontier caps, + per-query timeouts, and an honest `truncated` flag — a bad query + degrades to a flagged partial answer, never an unbounded scan and + never a silently wrong one. + +## The shape: one IR, two executors + +``` + GraphJQ text ─┐ + Cypher text ──┼─► frontend (pure, runs in path-cli) ─► query_ast IR + JSON AST ─────┘ │ + ┌─────────────────────────────┤ + ▼ ▼ + LOCAL: graphdb-engine REMOTE: POST IR (or text) + over GraphReader impl to Pathbase /query; + on the SQLite index server runs the same + (files stay canonical) engine over path_steps +``` + +The IR is the wire contract. `path query` parses whichever frontend +the user (or agent) wrote, then either executes locally or sends the +IR to Pathbase — same semantics document, same planner, same executor +code, so the same query provably means the same thing in both places. +Scope selection stays a CLI concern: local scope flags pick cache +docs; a Pathbase URL/`owner/repo` scope routes the IR remotely. + +### Local: `GraphReader` over the index + +This composes with `query-acceleration.md`'s Approach A rather than +competing with it. The SQLite index gains graph shape — either the +step rows plus a derived `edges` table, or facto's core-schema layout +(`vertices` / `edges` / `property_index`) directly. Files remain +canonical; the index remains disposable; the `GraphReader` impl is +~11 methods of SQLite reads. The engine's batched-neighbor access +pattern (`neighbor_ids_batch`) is exactly an indexed +`WHERE src_id IN (…) AND label = ?` scan. + +### Remote: `GraphReader` over what Pathbase already has + +Pathbase's schema converged on the same model independently: + +| facto core | pathbase today | +|---|---| +| `vertices` (JSONB props) | `path_steps.data` (JSONB, system of record) | +| `property_index` (hot scalars) | denormalized `actor_kind`, `change_type`, `timestamp` columns | +| `edges` | `step_parents` (the DAG), `graph_paths` (membership) | +| `has(k; p)` index pushdown | GIN `jsonb_path_ops` containment index | + +A server-side `GraphReader` over these tables is a mapping layer, not +a migration. Visibility enforcement stays where it lives today: the +reader impl applies `Visibility::can_read` scoping to every id/ +neighbor fetch, so the query engine can never see rows the REST/ +GraphQL surfaces wouldn't show ("one rule, one helper, no drift" +extends to the new surface). The fixed GraphQL resolvers +(`ancestors`, `nearestAncestorWith`, `toolUsage`) become one-line +queries, and new questions stop costing resolver code. + +## The property-graph mapping + +One graph per scope (locally: the whole cache; on Pathbase: a +repo/viewer scope — open question below). Vertex/edge vocabulary, +derived deterministically from toolpath documents at index time: + +**Vertices** +- `step` — props: the step body (actor, timestamp, intent, outcome, + token usage…), plus materialized `dead_end` (computed at index time + from head-ancestry, exactly as the jaq wrapper does today) and the + identity triple (`cache_id`, `path_id`, `step_id`). +- `path` — the session: `kind`, `source`, `base`, `title` + (first user message), token totals. +- `artifact` — a file/URL touched by any step, id = the normalized + artifact key. Shared across sessions — this vertex is what makes + cross-session file queries one hop instead of a scan. +- `actor`, `tool` — small dimension vertices (`human:ben`, + `tool:rustfmt`); optional but cheap, and they make grouping + traversals natural. + +**Edges** +- `parent`: step → step (the DAG; = `step_parents` on the server) +- `in_path`: step → path (membership) +- `head`: path → step +- `touches`: step → artifact (props: change perspective summary, + ± lines) +- `delegates`: step → path (sub-agent sessions) +- `invokes`: step → tool (= `step_tool_invocations`) + +**Example queries.** GraphJQ (curl-friendly, pipeline-shaped): + +``` +# Sessions that touched a file +V("artifact", id: "src/query/plan.rs") + | in("touches") | out("in_path") | dedup + | { id: id(), title: .title, source: .source } + +# Fork points: live ancestors of dead-end steps +V("step") | has("dead_end"; true) + | repeat(out("parent"); max_depth: 8) + | has("dead_end"; false) | dedup +``` + +Cypher (the same IR, the dialect agents already know): + +```cypher +MATCH (s:step)-[:touches]->(:artifact {id: "src/query/plan.rs"}), + (s)-[:in_path]->(p:path) +RETURN DISTINCT p.title, p.source + +MATCH (s:step)-[:in_path]->(p:path) +WHERE s.actor STARTS WITH "agent:" +RETURN p.title, sum(s.output_tokens) AS toks +ORDER BY toks DESC LIMIT 10 +``` + +Neither is expressible as *one* reasonable jaq program over the flat +wrapped-step array (the first needs a join through a shared artifact, +the second is fine in jaq — but the fork-point query is a bounded +transitive closure, which jaq simply doesn't have). + +## Would it be agent friendly? + +Yes — and more interestingly, the stack is agent-friendly in layers, +so different consumers can enter at different levels: + +1. **Cypher is the agent dialect.** openCypher is abundantly + represented in model training data (the Neo4j corpus); agents write + `MATCH … WHERE … RETURN` fluently, with none of the learning-curve + problem a novel language has. It is also facto's most-tested + frontend (261 test markers + a TCK harness). An agent-facing + `query` tool that accepts Cypher needs almost no prompt budget. +2. **GraphJQ is the human/shell dialect.** jq-flavored, curl-friendly, + pipeline-shaped — but effectively zero training-data presence, so + agents would need the language card in context. Fine for humans; + second choice for agents. +3. **The JSON AST is the structured-generation dialect.** An agent (or + a tool harness) can emit the IR directly under a JSON schema — + syntax errors become unrepresentable, and the harness can + constrain generation. This is the most reliable path for + programmatic agent use. +4. **The feedback loop is built for iteration.** `validate` never + 4xxs and returns parse errors with line/column/span; `compile` + returns the normalized AST and warnings; `explain` returns the + physical plan. That triple is exactly the self-correction loop + agentic query-writing wants — an agent can check before running, + and repair from structured errors. +5. **Bounded semantics protect both sides.** Depth clamps, frontier + caps, timeouts, and the honest `truncated` flag mean an agent's + bad query costs a bounded amount and a partial answer is *labeled* + partial — agents act on results without a human sanity-checking + each one, so "never silently wrong" matters more for them than + for interactive use. +6. **A schema card completes it.** Agents need the vertex/edge + vocabulary in context. `path kind` already prints field-reference + schemas; the graph vocabulary above is small enough to ship the + same way (`path kind graph` or similar). + +The honest limit: for pure step-content munging — reshaping the JSON +of steps you've already found — jaq remains stronger than GraphJQ's +projection language. The two surfaces are complementary: the graph +frontend normalizes *relational* queries and the local/remote split; +jaq stays for content transformation. (Agents know jq well too, so +the pairing costs little.) + +## Friction and open questions + +- **`graphdb-core` unconditionally depends on sqlx-core** (driverless) + and the engine is async (`async-trait`). For path-cli: tokio is + already in the tree (watcher features), so a small runtime shim + suffices. For the wasm playground: unvalidated; simplest is to gate + graph query off emscripten like `sync` already is. If the sqlx dep + bothers us, splitting it out of `graphdb-core` is a facto-side + cleanup (both projects are ours). +- **facto crates are `publish = false`.** Embedding means publishing + the four crates path-cli needs (`ir`, `core`, `engine`, + frontends) + or a git dependency. Ours to decide. +- **Tenancy mapping on Pathbase**: graph-per-repo keeps `graph_id` + natural but makes cross-repo queries a fan-out; a viewer-scoped + virtual graph inverts that. Needs its own pass. +- **`dead_end` and derived vertices** (`artifact`, `actor`, `tool`) + must be materialized identically at both index sites (local SQLite, + Pathbase ingest) or the "same query, same answer" promise breaks. + The derivation belongs in one shared place — plausibly a small + `toolpath-graph` crate defining the document → vertices/edges + mapping, used by both. +- **Two query surfaces in one CLI** need a coherent story: `path + query` (jaq, content) vs. a graph surface (`path query --graph` / + `path g`?) — naming and docs matter so agents and humans pick the + right tool without a decision tree. +- **facto is a proof-of-concept** by its own README (no auth, single + backend). Embedding the engine/IR crates is the low-risk slice — + the server, blobs, importers, and Wikidata layers stay behind. + +## Cheapest end-to-end validation + +Before committing to any of it: a spike that (1) implements +`GraphReader` over an in-memory graph built from one cached toolpath +doc (no SQLite work), (2) runs the two example queries above through +`graphdb_cypher::compile` + `graphdb_engine::traversal::execute`, and +(3) hands the same IR JSON to a copy of the engine mounted over a +pathbase-db test database. If that round-trips, the rest is schema +plumbing and product surface. diff --git a/docs/design/query-acceleration.md b/docs/design/query-acceleration.md new file mode 100644 index 00000000..dfd3cdba --- /dev/null +++ b/docs/design/query-acceleration.md @@ -0,0 +1,252 @@ +# Query acceleration for the local cache + +Status: design exploration, 2026-07. Nothing here is implemented. + +## Problem + +`path query` answers a jaq filter over every step in the local cache +(`~/.toolpath/documents/*.json`). The engine streams one document at a +time (`query/mod.rs::stream_files`), wraps each step in its source +context, and hands the planner (`query/plan.rs`) arrays it can often +stream or decompose instead of slurping. That bounds *memory* well, but +every query still pays the full parse cost of the cache: + +- read every doc file, +- serde-parse each into a `Graph`, +- re-serialize each step to `serde_json::Value` (`wrap_path`), +- convert to `jaq_json::Val`, +- run the filter. + +Measured baseline (2026-07-23, release build, M-series laptop): + +| Cache | Query | Wall time | +|---|---|---| +| 97 docs, 114 MB, 46,043 steps | `length` | ~0.9 s | +| same | `map(select(.dead_end)) \| length` | ~1.0 s | + +Almost all of it is user CPU — JSON parsing, not disk. Scope flags +(`--source`, `--project`, `--parent-dir`, `--kind`) don't help the +un-scoped case, and even scoped queries parse every doc before the +in-code filters discard most of them. + +The design question: what structure, maintained ahead of query time, +avoids the most parsing — without changing any answer? + +## Constraints (all approaches) + +- **Files stay canonical.** `~/.toolpath/documents/*.json` remains the + cache; any derived structure is an index that can be deleted and + rebuilt from the files at any time. `sync.json` stays the manifest — + it holds real state (evicted records, peeked cwds) that is not + rebuildable from the files. +- **The planner never changes an answer.** Whatever the accelerator + serves must equal the slurp path byte-for-byte, same guarantee (and + same test style) as `query/filter.rs` asserts for streaming today. +- **Output order is part of the answer**: docs in `list_cached` order + (file mtime, newest first), steps in document order. +- **wasm builds without it.** The playground has no rusqlite (it's in + path-cli's `cfg(not(target_os = "emscripten"))` dependency block); an + accelerator must be an optional fast path, not a dependency of + correctness. + +## Approach A: SQLite step-row index over the cache files + +### Shape + +One database, `$CONFIG_DIR/index.db`, holding every wrapped step as a +row. The row stores the step exactly as `wrap_path` emits it today — +`{"step": …, "change": …, "cache_id": …, "path": …, "dead_end": …}` — +in compact JSON, so index time absorbs the work queries currently +repeat: dead-end set computation, path-context cloning, wrapping, and +the double conversion (typed `Graph` → `Value` → `Val` becomes one +text → `Val` parse). + +```sql +PRAGMA journal_mode = WAL; +PRAGMA user_version = 1; + +-- One row per indexed cache file: the freshness stamp. +CREATE TABLE documents ( + cache_id TEXT PRIMARY KEY, + modified TEXT NOT NULL, -- file mtime at index time + size INTEGER NOT NULL, -- file size at index time + indexed_at TEXT NOT NULL +); + +-- One row per wrapped step. +CREATE TABLE steps ( + cache_id TEXT NOT NULL REFERENCES documents(cache_id) ON DELETE CASCADE, + seq INTEGER NOT NULL, -- position within the doc: output order + json TEXT NOT NULL, -- compact wrapped step + -- Hot fields, extracted by the schema itself so they can't drift: + step_id TEXT GENERATED ALWAYS AS (json_extract(json,'$.step.id')) VIRTUAL, + actor TEXT GENERATED ALWAYS AS (json_extract(json,'$.step.actor')) VIRTUAL, + ts TEXT GENERATED ALWAYS AS (json_extract(json,'$.step.timestamp')) VIRTUAL, + dead_end INT GENERATED ALWAYS AS (json_extract(json,'$.dead_end')) VIRTUAL, + source TEXT GENERATED ALWAYS AS (json_extract(json,'$.path.meta.source')) VIRTUAL, + kind TEXT GENERATED ALWAYS AS (json_extract(json,'$.path.meta.kind')) VIRTUAL, + base TEXT GENERATED ALWAYS AS (json_extract(json,'$.path.base.uri')) VIRTUAL, + PRIMARY KEY (cache_id, seq) +); +CREATE INDEX steps_actor ON steps(actor); +CREATE INDEX steps_dead_end ON steps(dead_end); +CREATE INDEX steps_source ON steps(source); +CREATE INDEX steps_ts ON steps(ts); +CREATE INDEX steps_base ON steps(base); +``` + +`VIRTUAL` generated columns cost no row storage; the indexes on them +store the extracted values — exactly where they're wanted. Scope flags +become `WHERE` clauses against these columns; no separate per-path +table is needed because every step row carries its path context. + +### Query execution: superset predicate pushdown + +Compiling arbitrary jaq to SQL is a correctness tarpit, and it's never +necessary. The pushdown layer only emits a **necessary condition** of +the filter — a `WHERE` clause keeping a *superset* of what the filter +keeps — because jaq still runs the real filter over every surviving +row. A superset predicate can fail to prune; it can never change an +answer. This is `plan.rs`'s existing ethos (conservative recognition, +slurp when unsure) extended one level down: + +- `map(select(.dead_end))` → `WHERE dead_end = 1`, residual jaq + unchanged on survivors. +- `map(select(.step.actor | startswith("agent:")))` → + `WHERE actor GLOB 'agent:*'`. +- Conjunctions push each recognized conjunct; unrecognized conjuncts + push nothing (the residual handles them). +- `map(select(any(.change[].structural.token_usage; …)))` → nothing + recognizable, push `TRUE`, scan all rows — today's behavior. + +A second, stricter tier: when the recognized predicate is provably +*equal* to the select body (not merely implied by it) and the tail is +one the planner already understands (`length`, top-N), the whole query +collapses into SQL — `SELECT count(*) FROM steps WHERE dead_end = 1` — +and no JSON is parsed at all. + +Expected effect at the measured cache size: + +| Tier | `map(select(.dead_end)) \| length` | +|---|---| +| today (parse everything) | ~1.2 s | +| step rows, no pushdown | ~0.4–0.6 s | +| superset pushdown + residual | ~0.2 s | +| exact absorption into SQL | ~5 ms | + +Rows stream `ORDER BY` doc-mtime-desc, `seq` — matching today's +ordering — into the existing planner unchanged. The byte-equality test +extends to the index path: index-served output must equal slurp output +exactly, over a seeded cache. + +### Freshness: the index is disposable + +- **Write-through**: `cache::write_cached` is the single funnel every + doc write already goes through (import, share, sync). It re-indexes + the doc in the same call: delete rows for `cache_id`, wrap once, + insert, stamp. +- **Query-time stat gate**: at query start, stat the cache files + against `documents` stamps (single-digit ms at ~100 files). Stale or + unknown docs re-index lazily; rows for deleted files are purged. + Out-of-band edits self-heal — same philosophy as sync's stat gate. +- **Deletable at any time**: removing `index.db` (or `p cache + reindex`) rebuilds from files at roughly the cost of one slurp query. + No migration ceremony; no "index disagrees with files" liability + beyond one stat pass. +- **wasm**: no rusqlite → no index → the slurp path runs as today. The + index is an accelerator, not a dependency. + +### Costs + +- **Storage roughly doubles.** The `json` column re-stores every + wrapped step: ~+130–150 MB beside the current 220 MB (compact JSON + vs. the pretty-printed files). A later "lean rows" variant could omit + `change` bodies and fall back to files when the filter touches + `.change` — deferred; it needs filter-touches-what analysis. +- **Sync gets slightly slower.** Indexing at write time is the + parse+wrap the doc would otherwise pay on first query; the cost moves + to where it amortizes. +- **A pushdown layer to maintain.** Each recognized predicate shape is + planner code with the same never-change-an-answer obligation the + streaming recognizer carries. Superset semantics keep mistakes + non-catastrophic (a wrong necessary-condition merely mis-prunes — + caught by the byte-equality tests — rather than silently changing + results, provided it stays a superset). + +### Implementation slices + +1. Index module + write-through + stat gate + ordered streaming into + the existing planner, byte-equality tests. (No speedup yet; locks in + correctness.) +2. Superset pushdown for `select` prefixes over the hot columns. +3. Exact absorption for recognized predicate + recognized tail. + +### Measured (implemented 2026-07-21, path-cli 0.17.0 + 0.18.0) + +Both this approach and the parallel scan landed; numbers on the real +97-doc / 114 MB / 46k-step cache (medians of 5, re-measured 2026-07-23 +after a 110 MB outlier document was deleted — see below), against the +pre-work baseline: + +| Query | baseline | rayon (0.17.0) | + index (0.18.0) | +|---|---|---|---| +| `length` (absorbed) | 898 ms | 292 ms | **30 ms** | +| `map(select(.dead_end)) \| length` (absorbed) | 957 ms | 309 ms | **30 ms** | +| `map(select(.step.actor \| startswith("agent:"))) \| length` | 976 ms | 310 ms | **169 ms** | +| `.[] \| select(.dead_end) \| .step.id` | 937 ms | 304 ms | 311 ms | +| `map(.step.actor) \| unique` (slurp) | 1021 ms | 856 ms | 881 ms | +| `group_by(.path.id) \| map(length) \| max` (slurp) | 1032 ms | 879 ms | 902 ms | + +Learnings vs. the estimates above: exact absorption delivered (~30×, +flat in cache size); predicated decomposes get ~6×; a predicated +*stream* lands at parity with the parallel scan — row-fetch + per-row +parse ≈ parallel whole-file parse — and unpredicated slurps and +streams are the parallel scan's territory (index rows within a few +percent). One-time index build: ~1.7 s; index size ~250 MB next to the +114 MB cache (rows + three hot-field indexes) — the "storage roughly +doubles" estimate held. The parse-only-parallelism tier estimated +above was measured and then superseded: per-file *filter execution* on +the workers (what 0.17.0 actually ships) reaches ~3.1× here and ~4.7× +on the even synthetic cache. + +The first round of measurements (2026-07-21) ran on a 220 MB version +of this same cache dominated by a single 110 MB document (48% of all +bytes, since deleted). Its effects were instructive: the parallel scan +was floored at ~2× (one document cannot split across cores), and the +predicated dead-end stream was *slower* than the scan (724 ms vs +615 ms) because its surviving rows carried that document's fat diffs. +Both effects vanished with the outlier — and both argue for the "lean +rows" variant sketched under Costs. + +## Other approaches (to explore) + +Sketches only; each gets its own section as it's worked through. + +- **Parallel scan, no index.** Keep the architecture exactly as-is and + parse docs on a thread pool. ~0.9 s is single-core; 8 cores → + ~200 ms with zero new state to maintain. Doesn't change asymptotics + and can't serve selective queries in ms, but it's the honest + baseline any index must beat — and it composes with every other + approach. +- **Binary sidecar cache.** Store each doc's pre-wrapped steps as a + fast binary serialization (e.g. postcard/bincode) beside the JSON, + invalidated by mtime. Kills the JSON-parse cost (the dominant term) + without SQL, pushdown, or a database; no help for selective queries + beyond the parse win. +- **Columnar / analytical engine.** DuckDB (or parquet + DataFusion) + over the step stream: vectorized scans make even full-cache + aggregations fast, and SQL becomes a user-facing query surface. Heavy + dependency; jq remains the compatibility contract, so it would sit + beside jaq, not replace it. +- **Result memoization.** Cache query outputs keyed by (filter text, + scope, index generation). Repeated queries become O(1); orthogonal to + everything above. +- **FTS5 topic search.** Not a jaq accelerator: a different query + modality ("find the session where I…") over user-prompt text / + titles. Pairs naturally with Approach A's database if chosen. + +A different axis entirely — normalizing local and Pathbase querying +under one graph query language (facto's GraphJQ/Cypher/SPARQL over a +shared IR, with the local SQLite index as one `GraphReader` backend) — +is explored in `graph-query-frontend.md`. It composes with Approach A +rather than competing with it. diff --git a/site/_data/crates.json b/site/_data/crates.json index 5ff8c063..93a3d5e6 100644 --- a/site/_data/crates.json +++ b/site/_data/crates.json @@ -113,7 +113,7 @@ }, { "name": "path-cli", - "version": "0.17.0", + "version": "0.18.0", "description": "Unified CLI (binary: path)", "docs": "https://docs.rs/path-cli", "crate": "https://crates.io/crates/path-cli",