diff --git a/CHANGELOG.md b/CHANGELOG.md index 25f6bc04..128900a3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,84 @@ 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 +plans that treat every file independently (`PerFileStream`, +`Decompose`), the whole per-file pipeline — read, parse, wrap, filter, +render — runs on rayon workers, with only ordered output assembly on +the main thread; `Slurp` plans parallelize the parse/wrap phase only +(the merged jaq array is `Rc`-based and must be built on one thread). +Output is byte-identical to the sequential scan — same ordering, same +warnings, same error precedence — enforced by tests pinning the +parallel building blocks against the sequential engine. + +Measured on a real 97-doc / 114 MB cache (M-series, medians of 5): +streamed and decomposed queries drop from ~0.95 s to ~0.31 s (~3.1×), +slurped queries improve ~1.2×. On an even 60-doc / 56 MB synthetic +cache the same queries run ~4.7× faster (e.g. `length` 966 ms → +206 ms). + +- **`path-cli`** (0.17.0): + - `query/mod.rs`: plan dispatch (`execute_plan`) and a chunked + parallel driver (`for_each_file`) that preserves selection order, + per-file warning order, and explicit-file error positions. + - `query/filter.rs`: per-file worker evaluation (`render_file`, + `partials_file` — partials cross threads as compact JSON bytes and + are reparsed on the consuming thread), a per-thread compiled-filter + cache, and `finish_decompose` shared by the sequential and parallel + drivers so the zero-file rule lives once. + - The emscripten (playground) build keeps the sequential engine — + no threads there. + ## `path p cache sync` — incremental session ingestion — 2026-07-16 Adds `path p cache sync [types…]`, the first step toward a cache that @@ -101,6 +179,20 @@ hand. and `--no-sync` opts out. This is the piece that makes the cache an implementation detail: a new user can run `path query` with no setup and get their sessions. + - `--parent-dir ` / `-d` on `p cache sync` and `path query` + restricts ingestion (and the query's reads) to artifacts under a + directory. Stat gate first, always: unchanged+cached artifacts skip + before any scope check; only artifacts that would cost a derive get + the constraint, with a one-line peek for codex and copilot (whose + cwd lives inside the session file) memoized into the manifest so it + happens at most once per artifact — and a derive never clobbers the + memoized peeked cwd. The copilot peek tolerates `session.start` + anywhere in the first lines and top-level cwd keys. Claude project + matching happens in slug space (its dir slugs are lossy), and pi + project scoping compares in its dir-encoded space so hyphenated + paths match. Out-of-scope artifacts are noted in the manifest + ("known, not materialized") but not derived, tallied separately + (`N out of scope`), and never touch a materialized record's stamp. - **`toolpath-codex`** (0.6.1, extends the bump below): `session_id_from_stem` is public — the trailing UUID of a rollout filename stem (or the whole stem when it has none), the same diff --git a/CLAUDE.md b/CLAUDE.md index 2de9f950..5089401e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -131,6 +131,7 @@ cargo run -p path-cli -- p cache ls cargo run -p path-cli -- p cache rm cargo run -p path-cli -- p cache sync # ingest new/changed sessions from every harness cargo run -p path-cli -- p cache sync claude codex # only these artifact types +cargo run -p path-cli -- p cache sync -d ~/work/proj # only artifacts under this directory # Inspect / analyze cargo run -p path-cli -- p render dot --input doc.json @@ -138,6 +139,7 @@ cargo run -p path-cli -- p render md --input doc.json --detail full # Query the whole local cache with a jaq (jq) filter over wrapped steps. # Queries auto-sync their scope first (--source claude syncs only claude; # --input-only queries never touch the cache); --no-sync opts out. +# --parent-dir/-d scopes both the sync and the read to that subtree. cargo run -p path-cli -- query 'map(select(.dead_end)) | map(.step.id)' cargo run -p path-cli -- query --source claude 'map(select(any(.change[].structural.token_usage; .input_tokens > 50000)))' cargo run -p path-cli -- query --input doc.json 'map(select(.step.actor | startswith("agent:")))' @@ -210,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`: 353 unit + 119 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` @@ -279,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. `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. -- 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 — 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 `p cache rm` evicts a doc (rm downgrades the record; the next sync re-materializes it, and sync also verifies the doc file actually exists before skipping, so even out-of-band deletions self-heal). 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. +- `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 6cdb1e62..f769fec1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -543,6 +543,31 @@ dependencies = [ "libc", ] +[[package]] +name = "crossbeam-deque" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5181e0de7b61eb03a81e347d6dd8797bae9da5146707b51077e2d71a54ec0ceb" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61803da095bee82a81bb1a452ecc25d3b2f1416d1897eb86430c6159ef717c17" + [[package]] name = "crossterm" version = "0.29.0" @@ -2441,7 +2466,7 @@ dependencies = [ [[package]] name = "path-cli" -version = "0.16.0" +version = "0.18.0" dependencies = [ "anyhow", "assert_cmd", @@ -2457,6 +2482,7 @@ dependencies = [ "pathbase-client", "predicates", "rand 0.9.4", + "rayon", "regex", "reqwest", "rusqlite", @@ -3032,6 +3058,26 @@ dependencies = [ "bitflags 2.11.1", ] +[[package]] +name = "rayon" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + [[package]] name = "recvmsg" version = "1.0.0" diff --git a/Cargo.toml b/Cargo.toml index 23223581..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.16.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"] } @@ -54,6 +54,7 @@ similar = "2" tempfile = "3.15" insta = "1" rusqlite = { version = "0.32", features = ["bundled"] } +rayon = "1.10" uuid = { version = "1", features = ["serde"] } [profile.wasm] diff --git a/crates/path-cli/Cargo.toml b/crates/path-cli/Cargo.toml index 575c972f..81d425ac 100644 --- a/crates/path-cli/Cargo.toml +++ b/crates/path-cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "path-cli" -version = "0.16.0" +version = "0.18.0" edition.workspace = true license.workspace = true repository = "https://github.com/empathic/toolpath" @@ -56,6 +56,7 @@ git2 = { workspace = true } reqwest = { workspace = true } tokio = { workspace = true } rusqlite = { workspace = true } +rayon = { workspace = true } uuid = { workspace = true, features = ["v4"] } # Embedded fuzzy picker — used when external `fzf` isn't on PATH. # Disable default features to avoid pulling in skim's CLI deps 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, + + /// Only ingest artifacts living under this directory (subtree + /// match). Out-of-scope artifacts are noted in the manifest but + /// not derived. + #[arg(long, short = 'd')] + parent_dir: Option, }, } @@ -39,7 +47,7 @@ pub fn run(op: CacheOp) -> Result<()> { CacheOp::Ls => run_ls(), CacheOp::Rm { id } => run_rm(&id), #[cfg(not(target_os = "emscripten"))] - CacheOp::Sync { types } => run_sync(types), + CacheOp::Sync { types, parent_dir } => run_sync(types, parent_dir), } } @@ -63,16 +71,20 @@ fn run_rm(id: &str) -> 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(()) } #[cfg(not(target_os = "emscripten"))] -fn run_sync(types: Vec) -> Result<()> { +fn run_sync(types: Vec, parent_dir: Option) -> Result<()> { let explicit = !types.is_empty(); let types = resolve_types(&types); let bundle = HarnessBundle::from_environment(); - let outcomes = sync_bundle(&bundle, &types, &mut Progress::new())?; + let outcomes = sync_bundle(&bundle, &types, parent_dir.as_deref(), &mut Progress::new())?; eprint!("{}", render_summary(&outcomes, explicit)); Ok(()) } @@ -190,6 +202,9 @@ fn render_summary(outcomes: &[(ArtifactType, SyncOutcome)], explicit: bool) -> S if o.failed > 0 { s.push_str(&format!(", {} failed", o.failed)); } + if o.out_of_scope > 0 { + s.push_str(&format!(", {} out of scope", o.out_of_scope)); + } s.push('\n'); } if s.is_empty() { @@ -238,6 +253,7 @@ mod tests { updated: 1, unchanged: 3, failed: 0, + out_of_scope: 0, }, ), (ArtifactType::Cursor, SyncOutcome::default()), @@ -259,6 +275,7 @@ mod tests { updated: 0, unchanged: 1, failed: 2, + out_of_scope: 0, }, )]; let s = render_summary(&outcomes, false); diff --git a/crates/path-cli/src/cmd_query.rs b/crates/path-cli/src/cmd_query.rs index b7b7df7b..38a0bd71 100644 --- a/crates/path-cli/src/cmd_query.rs +++ b/crates/path-cli/src/cmd_query.rs @@ -42,6 +42,11 @@ pub struct QueryArgs { #[arg(long)] project: Option, + /// Keep only paths whose base lives under this directory (subtree + /// match), and scope the implicit sync to it likewise. + #[arg(long, short = 'd')] + parent_dir: Option, + /// Keep only paths whose meta.kind matches this selector /// (semver prefix, e.g. `agent-coding-session` or `…/v1.0`). #[arg(long)] @@ -103,6 +108,7 @@ pub fn run(args: QueryArgs, pretty: bool) -> Result<()> { ids: args.ids, inputs: args.input, project: args.project, + parent_dir: args.parent_dir, kind: args.kind, }; @@ -123,7 +129,7 @@ fn sync_query_scope(args: &QueryArgs) { return; } let bundle = crate::harness::HarnessBundle::from_environment(); - match crate::sync::sync_bundle(&bundle, &types, &mut ()) { + match crate::sync::sync_bundle(&bundle, &types, args.parent_dir.as_deref(), &mut ()) { Ok(outcomes) => { for (t, o) in outcomes { if o.new + o.updated + o.failed > 0 { diff --git a/crates/path-cli/src/query/filter.rs b/crates/path-cli/src/query/filter.rs index d7a68a81..159fe520 100644 --- a/crates/path-cli/src/query/filter.rs +++ b/crates/path-cli/src/query/filter.rs @@ -59,7 +59,6 @@ pub fn execute( eval_print(&main, merged, out, &pp, raw)?; } Plan::Decompose { reduce } => { - let reducer = compile(reduce)?; let mut partials: Vec = Vec::new(); let mut saw_file = false; let mut emit = |val: Val| { @@ -68,23 +67,109 @@ pub fn execute( Ok(()) }; run_files(&mut emit)?; - if saw_file { - let merged: Val = partials.into_iter().collect(); - eval_print(&reducer, merged, out, &pp, raw)?; - } else { - // No document contributed a partial: the decomposition - // identity `reduce(⋃ main(fᵢ)) == main(⋃ fᵢ)` degenerates to - // `main([])`. Run the *main* filter over an empty array so the - // answer matches slurp (`length` → 0, `sort_by|.[:N]` → []), - // not the reducer over `[]` (which would give null / error). - let empty: Val = std::iter::empty().collect(); - eval_print(&main, empty, out, &pp, raw)?; - } + finish_decompose(main_src, reduce, partials, saw_file, compact, raw, out)?; } } Ok(()) } +/// Finish a `Decompose` plan over the gathered per-file partials — shared by +/// the sequential and parallel drivers so the zero-file rule lives once. +/// +/// With no document contributing a partial, the decomposition identity +/// `reduce(⋃ main(fᵢ)) == main(⋃ fᵢ)` degenerates to `main([])`. Run the +/// *main* filter over an empty array so the answer matches slurp (`length` +/// → 0, `sort_by|.[:N]` → []), not the reducer over `[]` (which would give +/// null / error). +pub fn finish_decompose( + main_src: &str, + reduce_src: &str, + partials: Vec, + saw_file: bool, + compact: bool, + raw: bool, + out: &mut dyn Write, +) -> Result<()> { + let pp = pretty(compact); + if saw_file { + let reducer = compile(reduce_src)?; + let merged: Val = partials.into_iter().collect(); + eval_print(&reducer, merged, out, &pp, raw) + } else { + let main = compile(main_src)?; + let empty: Val = std::iter::empty().collect(); + eval_print(&main, empty, out, &pp, raw) + } +} + +/// Surface a filter's load/compile errors without running it. The parallel +/// drivers call this before touching any file, preserving the sequential +/// path's error precedence (a bad filter beats a missing `--id`). +#[cfg(not(target_os = "emscripten"))] +pub fn compile_check(code: &str) -> Result<()> { + compile(code).map(|_| ()) +} + +/// Compile `code` at most once per thread. Rayon workers are long-lived, so +/// per-file evaluation pays one compile per pool thread, not per file. +#[cfg(not(target_os = "emscripten"))] +fn with_compiled(code: &str, f: impl FnOnce(&Program) -> Result) -> Result { + use std::cell::RefCell; + thread_local! { + static CACHE: RefCell> = const { RefCell::new(None) }; + } + CACHE.with(|cache| { + let mut cache = cache.borrow_mut(); + if !matches!(&*cache, Some((cached, _)) if cached == code) { + *cache = Some((code.to_string(), compile(code)?)); + } + let (_, prog) = cache.as_ref().expect("just populated"); + f(prog) + }) +} + +/// Evaluate `code` over one file's steps and render its outputs exactly as +/// [`execute`]'s streaming printer would. Runs on worker threads. +#[cfg(not(target_os = "emscripten"))] +pub fn render_file( + code: &str, + steps: Vec, + compact: bool, + raw: bool, +) -> Result> { + with_compiled(code, |prog| { + let mut buf = Vec::new(); + eval_print(prog, steps_to_val(steps)?, &mut buf, &pretty(compact), raw)?; + Ok(buf) + }) +} + +/// Evaluate `code` over one file's steps and pack the partial outputs into +/// one compact JSON array. Jaq values are `Rc`-based and thread-local, so +/// partials cross threads as bytes; [`unpack_partials`] reparses them on the +/// consuming thread. +#[cfg(not(target_os = "emscripten"))] +pub fn partials_file(code: &str, steps: Vec) -> Result> { + with_compiled(code, |prog| { + let vals = eval_collect(prog, steps_to_val(steps)?)?; + let arr: Val = vals.into_iter().collect(); + let mut buf = Vec::new(); + jaq_json::write::write(&mut buf, &pretty(true), 0, &arr) + .map_err(|e| anyhow!("internal: could not pack partials: {e}"))?; + Ok(buf) + }) +} + +/// Reparse one file's packed partials (the inverse of [`partials_file`]). +#[cfg(not(target_os = "emscripten"))] +pub fn unpack_partials(bytes: &[u8]) -> Result> { + match jaq_json::read::parse_single(bytes) { + Ok(Val::Arr(items)) => Ok(items.iter().cloned().collect()), + Ok(_) => Err(anyhow!("internal: packed partials were not an array")), + Err(e) => Err(anyhow!("internal: could not reparse partials: {e}")), + } +} + /// Convert one file's wrapped steps into a jaq array value. The byte buffer is /// per-file (bounded), so no whole-cache serialization is ever held. pub fn steps_to_val(steps: Vec) -> Result { @@ -402,4 +487,93 @@ mod tests { // filter. assert_slurps("map({id: .step.id}) | (sort_by(.id)) | .[:1]"); } + + // ── The parallel drivers' building blocks ───────────────────────── + // + // `render_file`/`partials_file` + `finish_decompose` are what the + // parallel per-file drivers run on worker threads. Their output must be + // byte-identical to the sequential engine, which is itself pinned to + // slurp above. + + fn file_steps(f: &serde_json::Value) -> Vec { + f.as_array().cloned().unwrap_or_default() + } + + #[test] + fn render_file_concat_matches_sequential_stream() { + for (compact, raw) in [(true, false), (false, false), (true, true)] { + for code in [ + ".[] | select(.dead_end)", + r#".[] | select(.step.actor | startswith("agent:")) | .step.id"#, + "map(select(.tokens > 40))", + ] { + let mut seq: Vec = Vec::new(); + execute(&Plan::PerFileStream, code, compact, raw, &mut seq, |emit| { + for f in &fixture() { + emit(steps_to_val(file_steps(f))?)?; + } + Ok(()) + }) + .unwrap(); + + let mut par: Vec = Vec::new(); + for f in &fixture() { + par.extend(render_file(code, file_steps(f), compact, raw).unwrap()); + } + assert_eq!( + String::from_utf8(par).unwrap(), + String::from_utf8(seq).unwrap(), + "parallel render must equal sequential for `{code}` (compact={compact}, raw={raw})" + ); + } + } + } + + #[test] + fn partials_roundtrip_matches_sequential_decompose() { + for code in [ + "length", + "map(select(.dead_end)) | length", + "sort_by(-.tokens) | .[:2]", + "map({id: .step.id, t: .tokens})", + ] { + let plan = crate::query::plan::analyze(code); + let Plan::Decompose { reduce } = &plan else { + panic!("`{code}` should decompose"); + }; + let seq = run_with(&plan, code, &fixture()); + + let mut partials: Vec = Vec::new(); + let mut saw_file = false; + for f in &fixture() { + saw_file = true; + let bytes = partials_file(code, file_steps(f)).unwrap(); + partials.extend(unpack_partials(&bytes).unwrap()); + } + let mut par: Vec = Vec::new(); + finish_decompose(code, reduce, partials, saw_file, true, false, &mut par).unwrap(); + assert_eq!( + String::from_utf8(par).unwrap(), + seq, + "parallel decompose must equal sequential for `{code}`" + ); + } + } + + #[test] + fn finish_decompose_zero_files_matches_slurp() { + for code in ["length", "sort_by(.tokens) | .[:2]"] { + let plan = crate::query::plan::analyze(code); + let Plan::Decompose { reduce } = &plan else { + panic!("`{code}` should decompose"); + }; + let mut par: Vec = Vec::new(); + finish_decompose(code, reduce, Vec::new(), false, true, false, &mut par).unwrap(); + let none: &[serde_json::Value] = &[]; + assert_eq!( + String::from_utf8(par).unwrap(), + run_with(&Plan::Slurp, code, none) + ); + } + } } 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 08e079db..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}; @@ -30,6 +32,9 @@ pub struct Scope { pub inputs: Vec, /// `--project`: keep only paths whose `base` resolves to this directory. pub project: Option, + /// `--parent-dir`: keep only paths whose `base` lives under this + /// directory (subtree match). + pub parent_dir: Option, /// `--kind`: keep only paths whose `meta.kind` matches this selector. pub kind: Option, } @@ -48,22 +53,282 @@ 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()); - filter::execute(&plan, code, compact, raw, &mut out, |emit| { - stream_files(scope, emit) - })?; + 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 +/// here. `Slurp` needs all steps as one jaq array on one thread, so it +/// parallelizes parsing only. +#[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, ctx, emit) + }), + plan::Plan::PerFileStream => { + filter::compile_check(code)?; + for_each_file( + scope, + ctx, + |src| filter::render_file(code, doc_steps(ctx, src)?, compact, raw), + |bytes| { + out.write_all(&bytes)?; + Ok(()) + }, + ) + } + plan::Plan::Decompose { reduce } => { + filter::compile_check(code)?; + let mut partials: Vec = Vec::new(); + let mut saw_file = false; + for_each_file( + scope, + ctx, + |src| filter::partials_file(code, doc_steps(ctx, src)?), + |bytes| { + saw_file = true; + partials.extend(filter::unpack_partials(&bytes)?); + Ok(()) + }, + )?; + filter::finish_decompose(code, reduce, partials, saw_file, compact, raw, out) + } + } +} + +/// The playground build has no threads (emscripten without pthreads): every +/// plan runs on the sequential engine. +#[cfg(target_os = "emscripten")] +fn execute_plan( + scope: &Scope, + ctx: &LoadCtx, + plan: &plan::Plan, + code: &str, + compact: bool, + raw: bool, + out: &mut dyn Write, +) -> Result<()> { + filter::execute(plan, code, compact, raw, out, |emit| { + stream_files(scope, ctx, emit) + }) +} + +/// Drive the per-file parallel plans: evaluate `per_file` for each selected, +/// scoped document on a thread pool in chunks, consuming products in +/// selection order on this thread. Chunking keeps output (and per-file +/// warnings) byte-identical to a sequential scan while holding at most one +/// chunk of products in memory. Explicit-file failures abort at the file's +/// position; a corrupt cache file skips with a warning, exactly like the +/// sequential scan. +#[cfg(not(target_os = "emscripten"))] +fn for_each_file( + scope: &Scope, + ctx: &LoadCtx, + per_file: impl Fn(&DocSource) -> Result + Sync, + mut consume: impl FnMut(T) -> Result<()>, +) -> Result<()> { + use rayon::prelude::*; + + 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(&per_file).collect(); + for (src, res) in batch.iter().zip(products) { + match res { + Ok(product) => consume(product)?, + Err(e) if src.explicit => { + return Err(e.context(format!("read {}", src.label()))); + } + Err(e) => eprintln!("warning: skipping {}: {e:#}", src.label()), + } + } + } + Ok(()) +} + /// Where a document came from, and the `cache_id` to stamp on its steps. struct DocSource { cache_id: String, @@ -72,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 { @@ -88,41 +358,82 @@ impl DocSource { } } -/// Stream each selected, scoped document to `emit` as a jaq array value, -/// one file at a time. Only one document's `Graph` and step values are alive -/// per iteration; the executor decides how to combine them. -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); - - for src in select_files(scope)? { - let graph = match read_source(&src) { - Ok(g) => g, - // An explicitly named document that won't read/parse is a hard - // error (a typo'd `--input`, bad stdin). A corrupt file merely - // encountered while scanning the cache is skipped with a warning. - Err(e) if src.explicit => { - return Err(e.context(format!("read {}", src.label()))); - } - Err(e) => { - eprintln!("warning: skipping {}: {e:#}", src.label()); - continue; +/// Stream each selected, scoped document to `emit` as a jaq array value, in +/// selection order. Reading and parsing — the dominant cost — runs on a +/// thread pool in chunks; conversion to `Val` and emission stay on this +/// 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, + 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| doc_steps(ctx, src)).collect(); + for (src, res) in batch.iter().zip(parsed) { + emit_wrapped(src, res, emit)?; } - }; - let mut steps = Vec::new(); - wrap_graph( - &src, - &graph, - kind_sel.as_ref(), - project.as_deref(), - &mut steps, - ); - drop(graph); - emit(filter::steps_to_val(steps)?)?; + } } + + // The playground build has no threads (emscripten without pthreads). + #[cfg(target_os = "emscripten")] + for src in &sources { + let res = doc_steps(ctx, src); + emit_wrapped(src, res, emit)?; + } + Ok(()) } +/// Read, parse, and wrap one document's steps. +fn load_and_wrap( + src: &DocSource, + kind_sel: Option<&KindSelector>, + project: Option<&FsPath>, + parent_dir: Option<&FsPath>, +) -> Result> { + let graph = read_source(src)?; + let mut steps = Vec::new(); + wrap_graph( + &src.cache_id, + &graph, + kind_sel, + project, + parent_dir, + &mut steps, + ); + Ok(steps) +} + +/// Emit one file's outcome. An explicitly named document that won't +/// read/parse is a hard error (a typo'd `--input`, bad stdin); a corrupt +/// file merely encountered while scanning the cache is skipped with a +/// warning. +fn emit_wrapped( + src: &DocSource, + res: Result>, + emit: &mut dyn FnMut(Val) -> Result<()>, +) -> Result<()> { + match res { + Ok(steps) => emit(filter::steps_to_val(steps)?), + Err(e) if src.explicit => Err(e.context(format!("read {}", src.label()))), + Err(e) => { + eprintln!("warning: skipping {}: {e:#}", src.label()); + Ok(()) + } + } +} + /// Resolve the scope's file-selection flags to a deterministic list of /// documents to load. /// @@ -166,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. @@ -184,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 @@ -193,6 +506,7 @@ fn select_files(scope: &Scope) -> Result> { cache_id: inp.clone(), location: SourceLoc::File(PathBuf::from(inp)), explicit: true, + indexed: false, }); } } @@ -227,10 +541,11 @@ 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>, + parent_dir: Option<&FsPath>, out: &mut Vec, ) { for entry in &graph.paths { @@ -249,13 +564,18 @@ fn wrap_graph( { continue; } + if let Some(dir) = parent_dir + && !path_matches_parent_dir(path, dir) + { + 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()) @@ -272,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( @@ -307,13 +627,17 @@ fn path_context(path: &Path) -> serde_json::Value { /// Whether a path's `base` resolves to `project` (a canonicalized directory). /// Only `file://` bases can match; VCS bases (`github:…`) never do. fn path_matches_project(path: &Path, project: &FsPath) -> bool { - let Some(base) = &path.path.base else { - return false; - }; - let Some(fs) = base.uri.strip_prefix("file://") else { - return false; - }; - canonicalize_or_self(FsPath::new(fs)) == project + base_fs_path(path).is_some_and(|p| p == project) +} + +fn path_matches_parent_dir(path: &Path, parent_dir: &FsPath) -> bool { + base_fs_path(path).is_some_and(|p| p.starts_with(parent_dir)) +} + +fn base_fs_path(path: &Path) -> Option { + let base = path.path.base.as_ref()?; + let fs = base.uri.strip_prefix("file://")?; + Some(canonicalize_or_self(FsPath::new(fs))) } fn canonicalize_or_self(p: &FsPath) -> PathBuf { @@ -325,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") @@ -366,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]; @@ -384,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() @@ -406,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, &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, &mut out); + wrap_graph("g", &graph, Some(&miss), None, None, &mut out); assert!(out.is_empty(), "non-matching kind drops the whole path"); } @@ -447,6 +763,7 @@ mod tests { ids: vec![], inputs: vec!["/tmp/some.json".to_string(), "-".to_string()], project: None, + parent_dir: None, kind: None, }; let files = select_files(&scope).unwrap(); 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/src/sync/engine.rs b/crates/path-cli/src/sync/engine.rs index e5b0242e..c6f0f0f2 100644 --- a/crates/path-cli/src/sync/engine.rs +++ b/crates/path-cli/src/sync/engine.rs @@ -6,7 +6,7 @@ use anyhow::{Context, Result, anyhow}; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use super::sources::{self, ArtifactSource}; use crate::artifact::{ArtifactRef, ArtifactType}; @@ -23,7 +23,8 @@ const MANIFEST_CHECKPOINT_EVERY_WRITES: usize = 10; /// What the manifest remembers about one known artifact. A record /// with a `cache_id` is materialized in the cache; one without is -/// merely known — evicted by `p cache rm`. +/// merely known — seen during an out-of-scope sync, or evicted by +/// `p cache rm`. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub(crate) struct SyncRecord { /// Filesystem path the artifact is keyed under: the project @@ -57,11 +58,13 @@ pub(crate) struct SyncOutcome { pub(crate) updated: usize, pub(crate) unchanged: usize, pub(crate) failed: usize, + /// Artifacts needing work that a `--parent-dir` constraint excluded. + pub(crate) out_of_scope: usize, } impl SyncOutcome { pub(crate) fn total(&self) -> usize { - self.new + self.updated + self.unchanged + self.failed + self.new + self.updated + self.unchanged + self.failed + self.out_of_scope } } @@ -70,7 +73,8 @@ impl SyncOutcome { /// sync logically with no UI pass `&mut ()`. Per artifact type: /// one `begin` once the stat pass has sized the pending work, one /// `failed` per derive error (before its `tick`), one `tick` per -/// finished derive, one `end` when the type's loop is done. +/// finished derive or scope skip, one `end` when the type's loop +/// is done. pub(crate) trait SyncObserver { fn begin(&mut self, _artifact_type: ArtifactType, _pending: usize) {} fn tick(&mut self) {} @@ -92,6 +96,7 @@ impl SyncObserver for () {} pub(crate) fn sync_bundle( bundle: &HarnessBundle, types: &[ArtifactType], + parent_dir: Option<&Path>, observer: &mut dyn SyncObserver, ) -> Result> { let manifest = load_manifest()?; @@ -104,7 +109,7 @@ pub(crate) fn sync_bundle( out.push((artifact_type, SyncOutcome::default())); continue; }; - let artifacts = source.enumerate(); + let artifacts = source.enumerate(parent_dir); let records = manifest .get(artifact_type.name()) .cloned() @@ -114,6 +119,7 @@ pub(crate) fn sync_bundle( artifact_type, &artifacts, &records, + parent_dir, observer, )?; out.push((artifact_type, outcome)); @@ -173,6 +179,7 @@ fn sync_artifacts( artifact_type: ArtifactType, artifacts: &[ArtifactRef], records: &BTreeMap, + parent_dir: Option<&Path>, observer: &mut dyn SyncObserver, ) -> Result { let mut outcome = SyncOutcome::default(); @@ -200,8 +207,48 @@ fn sync_artifacts( .or_default() .insert(artifact.id.clone(), record); }; - // An artifact without a path must not erase one an earlier - // run recorded. + // Scope gate: only artifacts that would cost a derive get the + // constraint check (with a bounded peek for codex/copilot, + // memoized in the record so it happens at most once per + // artifact). The source compares in its own key space — + // claude and pi keys are lossy dir encodings. + if let Some(parent_dir) = parent_dir { + let dir = artifact + .path + .clone() + .or_else(|| existing.and_then(|r| r.path.clone())) + .or_else(|| source.peek_dir(&artifact.id)); + let in_scope = dir + .as_deref() + .is_some_and(|d| source.in_scope(d, parent_dir)); + if !in_scope { + outcome.out_of_scope += 1; + // Remember what we learned — but never touch the stamp + // of a materialized record, or its staleness would be + // masked from the next in-scope sync. + if existing.is_none_or(|r| r.cache_id.is_none()) { + stage( + &mut writes, + SyncRecord { + path: dir, + cache_id: None, + modified: artifact.modified, + size: artifact.size, + synced_at: Utc::now(), + }, + ); + unflushed += 1; + } + observer.tick(); + if unflushed >= MANIFEST_CHECKPOINT_EVERY_WRITES { + flush_writes(&mut writes)?; + unflushed = 0; + } + continue; + } + } + // An artifact without a path must not erase one a previous + // pass peeked and memoized (codex/copilot cwd). let memoized_path = artifact .path .clone() @@ -522,7 +569,7 @@ mod tests { write_claude_session(home, "-test-project", "sess-aaa", "Add a feature"); let bundle = claude_bundle(home); let source = sources::source_for(&bundle, ArtifactType::Claude).unwrap(); - let artifacts = source.enumerate(); + let artifacts = source.enumerate(None); assert_eq!(artifacts.len(), 1); assert_eq!(artifacts[0].id, "sess-aaa"); assert_eq!(artifacts[0].path.as_deref(), Some("/test/project")); @@ -541,7 +588,7 @@ mod tests { write_claude_session(home, "-test-project", "sess-bbb", "Fix a bug"); let bundle = claude_bundle(home); - let outcomes = sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap(); + let outcomes = sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap(); assert_eq!(outcomes.len(), 1); let (_, first) = outcomes[0]; assert_eq!( @@ -565,7 +612,8 @@ mod tests { "cache doc must exist for {cache_id}" ); - let (_, second) = sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap()[0]; + let (_, second) = + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap()[0]; assert_eq!( (second.new, second.updated, second.unchanged, second.failed), (0, 0, 2, 0) @@ -578,7 +626,7 @@ mod tests { with_cfg(|home| { write_claude_session(home, "-test-project", "sess-aaa", "Add a feature"); let bundle = claude_bundle(home); - sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap(); + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap(); let cache_id = load_manifest().unwrap()["claude"]["sess-aaa"] .cache_id @@ -596,7 +644,8 @@ mod tests { body.push('\n'); std::fs::write(&file, body).unwrap(); - let (_, outcome) = sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap()[0]; + let (_, outcome) = + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap()[0]; assert_eq!( ( outcome.new, @@ -619,7 +668,7 @@ mod tests { write_claude_session(home, "-test-project", "sess-aaa", "Add a feature"); let bundle = claude_bundle(home); - let outcomes = sync_bundle(&bundle, &[ArtifactType::Codex], &mut ()).unwrap(); + let outcomes = sync_bundle(&bundle, &[ArtifactType::Codex], None, &mut ()).unwrap(); assert_eq!(outcomes[0].1, SyncOutcome::default()); assert!( load_manifest().unwrap().is_empty(), @@ -633,13 +682,14 @@ mod tests { with_cfg(|home| { write_claude_session(home, "-test-project", "sess-aaa", "Add a feature"); let bundle = claude_bundle(home); - sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap(); + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap(); // Losing the manifest (or a prior manual `p import`) leaves a // cache entry sync doesn't know about; re-syncing must // overwrite it, not die on the exists-check. std::fs::remove_file(manifest_path().unwrap()).unwrap(); - let (_, outcome) = sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap()[0]; + let (_, outcome) = + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap()[0]; assert_eq!((outcome.new, outcome.failed), (1, 0)); }); } @@ -650,7 +700,7 @@ mod tests { write_claude_session(home, "-test-project", "sess-aaa", "Add a feature"); let bundle = claude_bundle(home); let source = sources::source_for(&bundle, ArtifactType::Claude).unwrap(); - let mut artifacts = source.enumerate(); + let mut artifacts = source.enumerate(None); artifacts.push(make_ref(ArtifactType::Claude, "does-not-exist")); let outcome = sync_artifacts( @@ -658,6 +708,7 @@ mod tests { ArtifactType::Claude, &artifacts, &BTreeMap::new(), + None, &mut (), ) .unwrap(); @@ -676,7 +727,7 @@ mod tests { with_cfg(|home| { write_claude_session(home, "-test-project", "sess-aaa", "Add a feature"); let bundle = claude_bundle(home); - sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap(); + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap(); let cache_id = load_manifest().unwrap()["claude"]["sess-aaa"] .cache_id .clone() @@ -697,7 +748,8 @@ mod tests { ) .unwrap(); - let (_, outcome) = sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap()[0]; + let (_, outcome) = + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap()[0]; assert_eq!( (outcome.new, outcome.updated, outcome.unchanged), (0, 1, 0), @@ -714,7 +766,8 @@ mod tests { ); // And the grown chain settles: a third sync is a no-op. - let (_, again) = sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap()[0]; + let (_, again) = + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap()[0]; assert_eq!((again.updated, again.unchanged), (0, 1)); }); } @@ -724,7 +777,7 @@ mod tests { with_cfg(|home| { write_claude_session(home, "-test-project", "sess-aaa", "Add a feature"); let bundle = claude_bundle(home); - sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap(); + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap(); // A record whose stamps are all None (stat failed when it // was written) must not match a stub whose stat also @@ -740,6 +793,7 @@ mod tests { ArtifactType::Claude, &[artifact], &records, + None, &mut (), ) .unwrap(); @@ -768,7 +822,8 @@ mod tests { record_artifact(artifact, &derived.cache_id).unwrap(); // The import's stamp must match sync's own enumeration. - let (_, outcome) = sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap()[0]; + let (_, outcome) = + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap()[0]; assert_eq!( ( outcome.new, @@ -781,12 +836,155 @@ mod tests { }); } + #[test] + fn parent_dir_scopes_path_keyed_enumeration() { + with_cfg(|home| { + write_claude_session(home, "-scope-alpha", "aaaa1111-x", "In alpha"); + write_claude_session(home, "-scope-beta", "bbbb2222-x", "In beta"); + let bundle = claude_bundle(home); + + let (_, scoped) = sync_bundle( + &bundle, + &[ArtifactType::Claude], + Some(Path::new("/scope/alpha")), + &mut (), + ) + .unwrap()[0]; + assert_eq!((scoped.new, scoped.out_of_scope), (1, 0)); + let manifest = load_manifest().unwrap(); + assert!( + !manifest["claude"].contains_key("bbbb2222-x"), + "pruned projects must not be enumerated or recorded" + ); + + // Unscoped sync picks up the rest. + let (_, full) = + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap()[0]; + assert_eq!((full.new, full.unchanged), (1, 1)); + }); + } + + fn codex_bundle(home: &Path, cwd: &str) -> HarnessBundle { + let codex_dir = home.join(".codex"); + let dir = codex_dir.join("sessions/2026/05/07"); + std::fs::create_dir_all(&dir).unwrap(); + let meta = format!( + r#"{{"timestamp":"2026-05-07T00:00:00Z","type":"session_meta","payload":{{"id":"00000000-0000-0000-0000-0000000000aa","timestamp":"2026-05-07T00:00:00Z","cwd":"{cwd}","originator":"codex-tui","cli_version":"test","source":"cli","model_provider":"openai"}}}}"# + ); + let user = r#"{"timestamp":"2026-05-07T00:00:01Z","type":"response_item","payload":{"type":"message","role":"user","content":[{"type":"input_text","text":"hi"}]}}"#; + std::fs::write( + dir.join("rollout-2026-05-07T00-00-00-00000000-0000-0000-0000-0000000000aa.jsonl"), + format!("{meta}\n{user}\n"), + ) + .unwrap(); + let resolver = toolpath_codex::PathResolver::new().with_codex_dir(&codex_dir); + HarnessBundle { + codex: Some(toolpath_codex::CodexConvo::with_resolver(resolver)), + ..Default::default() + } + } + + #[test] + fn out_of_scope_codex_peek_is_memoized_then_scope_match_derives() { + with_cfg(|home| { + let bundle = codex_bundle(home, "/work/proj"); + + // cwd lives outside the constraint: one bounded peek, a + // known-but-uncached record, no derive. + let (_, out) = sync_bundle( + &bundle, + &[ArtifactType::Codex], + Some(Path::new("/elsewhere")), + &mut (), + ) + .unwrap()[0]; + assert_eq!((out.new, out.out_of_scope), (0, 1)); + let rec = + load_manifest().unwrap()["codex"]["00000000-0000-0000-0000-0000000000aa"].clone(); + assert_eq!( + rec.path.as_deref(), + Some("/work/proj"), + "peeked cwd memoized" + ); + assert!(rec.cache_id.is_none(), "known, not materialized"); + + // Matching constraint: the memoized record answers the scope + // question and the artifact derives. + let (_, hit) = sync_bundle( + &bundle, + &[ArtifactType::Codex], + Some(Path::new("/work/proj")), + &mut (), + ) + .unwrap()[0]; + assert_eq!((hit.new, hit.updated, hit.out_of_scope), (0, 1, 0)); + let rec = + load_manifest().unwrap()["codex"]["00000000-0000-0000-0000-0000000000aa"].clone(); + assert!(rec.cache_id.is_some(), "materialized now"); + assert_eq!( + rec.path.as_deref(), + Some("/work/proj"), + "deriving must not clobber the memoized peeked cwd" + ); + }); + } + + fn copilot_bundle(home: &Path, id: &str, cwd: &str) -> HarnessBundle { + let copilot_dir = home.join(".copilot"); + let dir = copilot_dir.join("session-state").join(id); + std::fs::create_dir_all(&dir).unwrap(); + let start = format!( + r#"{{"type":"session.start","timestamp":"2026-07-01T00:00:00Z","data":{{"copilotVersion":"1.0.67","context":{{"cwd":"{cwd}"}}}}}}"# + ); + let user = + r#"{"type":"user.message","timestamp":"2026-07-01T00:00:01Z","data":{"content":"hi"}}"#; + std::fs::write(dir.join("events.jsonl"), format!("{start}\n{user}\n")).unwrap(); + let resolver = toolpath_copilot::PathResolver::new().with_copilot_dir(&copilot_dir); + HarnessBundle { + copilot: Some(toolpath_copilot::CopilotConvo::with_resolver(resolver)), + ..Default::default() + } + } + + #[test] + fn copilot_syncs_and_scopes_via_memoized_peek() { + with_cfg(|home| { + let bundle = copilot_bundle(home, "sess-cp", "/work/proj"); + + // Out-of-scope first: one peek, a known record with the cwd. + let (_, out) = sync_bundle( + &bundle, + &[ArtifactType::Copilot], + Some(Path::new("/elsewhere")), + &mut (), + ) + .unwrap()[0]; + assert_eq!((out.new, out.out_of_scope), (0, 1)); + let rec = load_manifest().unwrap()["copilot"]["sess-cp"].clone(); + assert_eq!(rec.path.as_deref(), Some("/work/proj")); + assert!(rec.cache_id.is_none()); + + // In scope: derives; then a plain re-sync is a no-op. + let (_, hit) = sync_bundle( + &bundle, + &[ArtifactType::Copilot], + Some(Path::new("/work")), + &mut (), + ) + .unwrap()[0]; + assert_eq!((hit.updated, hit.out_of_scope), (1, 0)); + let (_, again) = + sync_bundle(&bundle, &[ArtifactType::Copilot], None, &mut ()).unwrap()[0]; + assert_eq!(again.unchanged, 1); + }); + } + #[test] fn evicted_cache_entry_rematerializes_on_next_sync() { with_cfg(|home| { write_claude_session(home, "-test-project", "sess-aaa", "Add a feature"); let bundle = claude_bundle(home); - sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap(); + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap(); let cache_id = load_manifest().unwrap()["claude"]["sess-aaa"] .cache_id .clone() @@ -801,7 +999,8 @@ mod tests { .is_none() ); - let (_, outcome) = sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap()[0]; + let (_, outcome) = + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap()[0]; assert_eq!((outcome.new, outcome.updated), (0, 1)); assert!( crate::cache::cache_path(&cache_id).unwrap().exists(), @@ -815,7 +1014,7 @@ mod tests { with_cfg(|home| { write_claude_session(home, "-test-project", "sess-aaa", "Add a feature"); let bundle = claude_bundle(home); - sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap(); + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap(); let cache_id = load_manifest().unwrap()["claude"]["sess-aaa"] .cache_id .clone() @@ -825,7 +1024,8 @@ mod tests { // materialization, but sync verifies the doc exists. let doc = crate::cache::cache_path(&cache_id).unwrap(); std::fs::remove_file(&doc).unwrap(); - let (_, outcome) = sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap()[0]; + let (_, outcome) = + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap()[0]; assert_eq!((outcome.new, outcome.updated), (0, 1)); assert!(doc.exists()); }); @@ -848,7 +1048,7 @@ mod tests { .is_none() ); - sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap(); + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap(); let cache_id = fresh_cache_id( &bundle, ArtifactType::Claude, @@ -874,7 +1074,7 @@ mod tests { ) .is_none() ); - sync_bundle(&bundle, &[ArtifactType::Claude], &mut ()).unwrap(); + sync_bundle(&bundle, &[ArtifactType::Claude], None, &mut ()).unwrap(); assert!( fresh_cache_id( &bundle, @@ -900,6 +1100,42 @@ mod tests { }); } + #[test] + fn copilot_peek_accepts_top_level_cwd() { + with_cfg(|home| { + // Older CLIs store cwd at the payload top level, no + // `context` object — the peek must still find it. + let copilot_dir = home.join(".copilot"); + let dir = copilot_dir.join("session-state").join("sess-legacy"); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write( + dir.join("events.jsonl"), + concat!( + r#"{"type":"session.start","data":{"cwd":"/work/proj"}}"#, + "\n", + r#"{"type":"user.message","data":{"content":"hi"}}"#, + "\n" + ), + ) + .unwrap(); + let resolver = toolpath_copilot::PathResolver::new().with_copilot_dir(&copilot_dir); + let bundle = HarnessBundle { + copilot: Some(toolpath_copilot::CopilotConvo::with_resolver(resolver)), + ..Default::default() + }; + let (_, out) = sync_bundle( + &bundle, + &[ArtifactType::Copilot], + Some(Path::new("/elsewhere")), + &mut (), + ) + .unwrap()[0]; + assert_eq!(out.out_of_scope, 1); + let rec = load_manifest().unwrap()["copilot"]["sess-legacy"].clone(); + assert_eq!(rec.path.as_deref(), Some("/work/proj")); + }); + } + #[test] fn newest_first_orders_by_mtime_with_unstamped_last() { let mut old = make_ref(ArtifactType::Claude, "old"); diff --git a/crates/path-cli/src/sync/sources.rs b/crates/path-cli/src/sync/sources.rs index ba76608c..4ae87617 100644 --- a/crates/path-cli/src/sync/sources.rs +++ b/crates/path-cli/src/sync/sources.rs @@ -9,6 +9,7 @@ use anyhow::{Result, anyhow}; use chrono::{DateTime, Utc}; +use std::path::{Path, PathBuf}; use crate::artifact::{ArtifactRef, ArtifactType, claude_chain_stamp, stat_stamp}; use crate::derive::{self, DerivedDoc}; @@ -25,9 +26,10 @@ pub(crate) type Stamp = (Option>, Option); /// One provider, as the sync engine sees it. pub(crate) trait ArtifactSource { /// Enumerate this provider's artifacts with stat-level - /// fingerprints. Never reads session bodies. Listing errors warn - /// and skip so one broken provider can't block a run. - fn enumerate(&self) -> Vec; + /// fingerprints, pruning to `parent_dir` when the provider's + /// listing is path-keyed. Never reads session bodies. Listing + /// errors warn and skip so one broken provider can't block a run. + fn enumerate(&self, parent_dir: Option<&Path>) -> Vec; /// Stat-level fingerprint for a single artifact, resolved /// directly — the same stat targets as [`Self::enumerate`], so the @@ -39,6 +41,21 @@ pub(crate) trait ArtifactSource { /// through the same manager it was enumerated from, so listing and /// derivation always agree on provider roots. fn derive(&self, artifact: &ArtifactRef) -> Result; + + /// Bounded peek at the directory an artifact lives in, for + /// providers whose cheap listing doesn't carry it (the cwd lives + /// inside the session file). The engine memoizes the answer into + /// the manifest record, so this runs at most once per artifact. + fn peek_dir(&self, _id: &str) -> Option { + None + } + + /// Whether `dir` lies under `parent_dir` in this provider's key + /// space. Providers whose keys are lossy dir encodings (claude, + /// pi) compare in the encoded space instead of as real paths. + fn in_scope(&self, dir: &str, parent_dir: &Path) -> bool { + dir_in_scope(dir, parent_dir) + } } /// The source for `t`'s artifacts, borrowing its manager from @@ -69,6 +86,18 @@ fn require_path(artifact: &ArtifactRef) -> Result<&str> { .ok_or_else(|| anyhow!("artifact {} has no path", artifact.id)) } +fn canonicalize_or_self(p: &Path) -> PathBuf { + std::fs::canonicalize(p).unwrap_or_else(|_| p.to_path_buf()) +} + +/// Subtree check for a real filesystem path. Canonicalizes both +/// sides, but also accepts the raw parent so a not-yet-resolvable +/// constraint (or an unresolvable dir) still matches literally. +fn dir_in_scope(dir: &str, parent_dir: &Path) -> bool { + let d = canonicalize_or_self(Path::new(dir)); + d.starts_with(canonicalize_or_self(parent_dir)) || d.starts_with(parent_dir) +} + // ── claude ───────────────────────────────────────────────────────── struct ClaudeSource<'a>(&'a toolpath_claude::ClaudeConvo); @@ -79,7 +108,7 @@ impl ArtifactSource for ClaudeSource<'_> { /// segment and its id is rotation-stable; the fingerprint covers /// the whole chain (see `claude_chain_stamp`) because appends land /// in the newest segment, not the head file. - fn enumerate(&self) -> Vec { + fn enumerate(&self, parent_dir: Option<&Path>) -> Vec { let mut out = Vec::new(); let projects = match self.0.list_projects() { Ok(ps) => ps, @@ -90,6 +119,11 @@ impl ArtifactSource for ClaudeSource<'_> { } }; for project in projects { + if let Some(parent_dir) = parent_dir + && !claude_project_in_scope(&project, parent_dir) + { + continue; + } let heads = match self.0.list_conversations(&project) { Ok(h) => h, Err(e) => { @@ -118,6 +152,27 @@ impl ArtifactSource for ClaudeSource<'_> { fn derive(&self, artifact: &ArtifactRef) -> Result { derive::derive_claude_session_with(self.0, require_path(artifact)?, &artifact.id) } + + fn in_scope(&self, dir: &str, parent_dir: &Path) -> bool { + claude_project_in_scope(dir, parent_dir) + } +} + +/// Claude's project-dir slugs are lossy — '/', '_', and '.' all +/// became '-', and un-sanitizing only restores '/'. Comparing real +/// paths therefore misfilters any project containing '.' or '_'; +/// compare in slug space instead, where '/' boundaries are '-'. +fn claude_project_in_scope(project: &str, parent_dir: &Path) -> bool { + fn slug(s: &str) -> String { + s.replace(['/', '_', '.'], "-") + } + let p = slug(project); + [ + slug(&parent_dir.to_string_lossy()), + slug(&canonicalize_or_self(parent_dir).to_string_lossy()), + ] + .iter() + .any(|parent| p == *parent || p.starts_with(&format!("{parent}-"))) } // ── gemini ───────────────────────────────────────────────────────── @@ -128,7 +183,7 @@ impl ArtifactSource for GeminiSource<'_> { /// Session entries via a bounded identity peek (`toolpath-gemini` /// reads at most the first 4 KiB of a main file); the fingerprint /// stats the main file (or the orphan sub-agent directory). - fn enumerate(&self) -> Vec { + fn enumerate(&self, parent_dir: Option<&Path>) -> Vec { let mut out = Vec::new(); let projects = match self.0.list_projects() { Ok(ps) => ps, @@ -139,6 +194,11 @@ impl ArtifactSource for GeminiSource<'_> { } }; for project in projects { + if let Some(parent_dir) = parent_dir + && !dir_in_scope(&project, parent_dir) + { + continue; + } let entries = match self.0.resolver().list_session_entries(&project) { Ok(entries) => entries, Err(e) => { @@ -181,7 +241,7 @@ impl ArtifactSource for CodexSource<'_> { /// Rollout files, stat-only. The artifact id is the trailing UUID of /// the filename stem (`rollout--`); `read_session` /// accepts either the UUID or the full stem, so the fallback is safe. - fn enumerate(&self) -> Vec { + fn enumerate(&self, _parent_dir: Option<&Path>) -> Vec { let mut out = Vec::new(); let files = match self.0.io().list_rollout_files() { Ok(f) => f, @@ -216,6 +276,18 @@ impl ArtifactSource for CodexSource<'_> { fn derive(&self, artifact: &ArtifactRef) -> Result { derive::derive_codex_session_with(self.0, &artifact.id) } + + /// The rollout's first line is `session_meta`; its payload carries + /// cwd directly — one bounded read. + fn peek_dir(&self, id: &str) -> Option { + use std::io::{BufRead, BufReader}; + let file = self.0.resolver().find_rollout_file(id).ok()?; + let f = std::fs::File::open(file).ok()?; + let mut line = String::new(); + BufReader::new(f).read_line(&mut line).ok()?; + let v: serde_json::Value = serde_json::from_str(&line).ok()?; + Some(v.get("payload")?.get("cwd")?.as_str()?.to_string()) + } } // ── opencode ─────────────────────────────────────────────────────── @@ -225,7 +297,7 @@ struct OpencodeSource<'a>(&'a toolpath_opencode::OpencodeConvo); impl ArtifactSource for OpencodeSource<'_> { /// One header-only `SELECT` — `time_updated` is the fingerprint; no /// message bodies are loaded. - fn enumerate(&self) -> Vec { + fn enumerate(&self, _parent_dir: Option<&Path>) -> Vec { let mut out = Vec::new(); let sessions = match self.0.io().list_sessions(None) { Ok(s) => s, @@ -267,7 +339,7 @@ impl ArtifactSource for CursorSource<'_> { /// check) — `lastUpdatedAt` is the fingerprint. Bubble-less drafts /// are skipped; unlike `share`, composers without a workspace are /// included, since sync doesn't need to rank them by project. - fn enumerate(&self) -> Vec { + fn enumerate(&self, _parent_dir: Option<&Path>) -> Vec { let mut out = Vec::new(); let listings = match self.0.io().list_composers() { Ok(l) => l, @@ -314,7 +386,7 @@ impl ArtifactSource for PiSource<'_> { /// Session files stat-only; the id comes from a one-line header /// peek, falling back to the filename stem's `_` /// shape — the same resolution `read_session` accepts. - fn enumerate(&self) -> Vec { + fn enumerate(&self, parent_dir: Option<&Path>) -> Vec { let mut out = Vec::new(); let projects = match self.0.list_projects() { Ok(ps) => ps, @@ -325,6 +397,11 @@ impl ArtifactSource for PiSource<'_> { } }; for project in projects { + if let Some(parent_dir) = parent_dir + && !pi_project_in_scope(&project, parent_dir) + { + continue; + } let files = match toolpath_pi::reader::list_session_files(self.0.resolver(), &project) { Ok(f) => f, Err(e) => { @@ -373,6 +450,27 @@ impl ArtifactSource for PiSource<'_> { fn derive(&self, artifact: &ArtifactRef) -> Result { derive::derive_pi_session_with(self.0, require_path(artifact)?, &artifact.id) } + + fn in_scope(&self, dir: &str, parent_dir: &Path) -> bool { + pi_project_in_scope(dir, parent_dir) + } +} + +/// Pi's session-dir names encode '/' as '-', and decoding restores +/// every '-' to '/', so a decoded project string is wrong for any +/// real path containing a hyphen. Compare in the encoded space, +/// where '/' boundaries are '-'. +fn pi_project_in_scope(project: &str, parent_dir: &Path) -> bool { + fn enc(s: &str) -> String { + s.replace('/', "-") + } + let p = enc(project); + [ + enc(&parent_dir.to_string_lossy()), + enc(&canonicalize_or_self(parent_dir).to_string_lossy()), + ] + .iter() + .any(|parent| p == *parent || p.starts_with(&format!("{parent}-"))) } // ── copilot ──────────────────────────────────────────────────────── @@ -384,7 +482,7 @@ impl ArtifactSource for CopilotSource<'_> { /// `/events.jsonl` under `session-state/` (or its legacy /// sibling); the directory name is the id and the events file is /// the fingerprint target. - fn enumerate(&self) -> Vec { + fn enumerate(&self, _parent_dir: Option<&Path>) -> Vec { let mut out = Vec::new(); let mut seen = std::collections::HashSet::new(); let dirs = [ @@ -424,6 +522,30 @@ impl ArtifactSource for CopilotSource<'_> { fn derive(&self, artifact: &ArtifactRef) -> Result { derive::derive_copilot_session_with(self.0, &artifact.id) } + + /// `session.start` is usually first but the format tolerates it + /// appearing later, and older CLIs store cwd at the top of the + /// payload instead of under `context` — scan a bounded number of + /// lines and accept both shapes, mirroring toolpath-copilot's own + /// tolerance. + fn peek_dir(&self, id: &str) -> Option { + use std::io::{BufRead, BufReader}; + let file = self.0.resolver().events_file(id).ok()?; + let f = std::fs::File::open(file).ok()?; + BufReader::new(f).lines().take(10).find_map(|line| { + let v: serde_json::Value = serde_json::from_str(&line.ok()?).ok()?; + ["data", "payload"].iter().find_map(|env| { + let payload = v.get(env)?; + [payload.get("context").unwrap_or(payload), payload] + .iter() + .find_map(|scope| { + ["cwd", "workingDirectory", "working_dir"] + .iter() + .find_map(|k| Some(scope.get(k)?.as_str()?.to_string())) + }) + }) + }) + } } #[cfg(test)] @@ -446,4 +568,23 @@ mod tests { "git is recorded by `p import`, never enumerated or derived by sync" ); } + + #[test] + fn pi_scope_matching_survives_hyphenated_paths() { + // decode_project turns the dir-encoded '-' back into '/', so a + // real path /Users/b/my-repo arrives as /Users/b/my/repo; the + // comparator must match it against the real constraint anyway. + assert!(pi_project_in_scope( + "/Users/b/my/repo", + Path::new("/Users/b/my-repo") + )); + assert!(pi_project_in_scope( + "/Users/b/my/repo/sub", + Path::new("/Users/b/my-repo") + )); + assert!(!pi_project_in_scope( + "/Users/b/other", + Path::new("/Users/b/my-repo") + )); + } } diff --git a/crates/path-cli/tests/integration.rs b/crates/path-cli/tests/integration.rs index f5c22b36..836cf445 100644 --- a/crates/path-cli/tests/integration.rs +++ b/crates/path-cli/tests/integration.rs @@ -903,6 +903,30 @@ fn share_records_manifest_so_sync_skips() { .stderr(predicate::str::contains("0 new, 0 updated, 1 unchanged")); } +#[test] +fn cache_sync_parent_dir_limits_ingestion() { + let (home, _session_file) = claude_home_fixture(); + let cfg = tempfile::tempdir().unwrap(); + let project = home.path().join("proj"); + + cmd() + .env("HOME", home.path()) + .env("TOOLPATH_CONFIG_DIR", cfg.path()) + .args(["p", "cache", "sync", "claude", "-d", "/nowhere"]) + .assert() + .success() + .stderr(predicate::str::contains("0 new, 0 updated, 0 unchanged")); + + cmd() + .env("HOME", home.path()) + .env("TOOLPATH_CONFIG_DIR", cfg.path()) + .args(["p", "cache", "sync", "claude", "-d"]) + .arg(&project) + .assert() + .success() + .stderr(predicate::str::contains("1 new, 0 updated, 0 unchanged")); +} + #[test] fn share_uploads_cached_doc_when_source_unchanged() { let (home, session_file) = claude_home_fixture(); diff --git a/crates/path-cli/tests/query.rs b/crates/path-cli/tests/query.rs index 5ebbf029..43f68b81 100644 --- a/crates/path-cli/tests/query.rs +++ b/crates/path-cli/tests/query.rs @@ -600,3 +600,124 @@ fn input_only_query_never_touches_the_cache() { .stdout(predicate::str::contains("true")); assert!(!cfg.path().join("manifest.json").exists()); } + +#[test] +fn query_parent_dir_scopes_sync_and_read() { + let home = claude_home(); + let cfg = tempfile::tempdir().unwrap(); + let project = home.path().join("proj"); + + // Constraint matches the session's project: ingested and read. + fixture_query( + &home, + cfg.path(), + ["--parent-dir", project.to_str().unwrap(), "length > 0"], + ) + .success() + .stdout(predicate::str::contains("true")) + .stderr(predicate::str::contains("synced claude: 1 new")); + + // Constraint elsewhere: nothing read, nothing new ingested. + fixture_query(&home, cfg.path(), ["--parent-dir", "/nowhere", "length"]) + .success() + .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 99bcad0b..93a3d5e6 100644 --- a/site/_data/crates.json +++ b/site/_data/crates.json @@ -113,7 +113,7 @@ }, { "name": "path-cli", - "version": "0.16.0", + "version": "0.18.0", "description": "Unified CLI (binary: path)", "docs": "https://docs.rs/path-cli", "crate": "https://crates.io/crates/path-cli",