diff --git a/VIBES.org b/VIBES.org
index a843621..c745922 100644
--- a/VIBES.org
+++ b/VIBES.org
@@ -8837,11 +8837,106 @@ Tests should not rely on the repo config.
 :END:
 
 Add a top-level config section for tasks commands, name it =tasks=. Move the current top-level =capture= config into =tasks.capture=. Define a new section for the =tasks.list= that will contain all of its configuration & defaults.
-*** TODO Streamline query execution
+*** REWORKED Streamline query execution
+:LOGBOOK:
+- State "REWORKED"   from "TODO"       [2026-06-23 Tue 20:10]
+:END:
 The commands use a common ETL pipeline where they load some data from files, apply filters/ordering/mutators and lastly present the results in some way. Sometimes the documents are then saved.
-It would be nice to make query execution as streamlined as possible, ideally, =org.logic.query= would expose a function =run_query= that takes a list of documents, a list of query stages to run and the context for the query to run in, and compiles and executes the query and returns the result.
+It would be nice to make query execution as streamlined as possible, ideally, =org.query.runner= would expose a function =run_query= that takes a list of documents, a list of query stages to run and the context for the query to run in, and compiles and executes the query and returns the result.
 That streamlined query execution should be used in all places that extract or transform the org document data - the data should be loaded, then processed with the =run_query= function and used as needed.
 Error handling should be left to the caller (=run_query= should not accept customizable error types/builders).
+**** DONE src/org/commands/agenda/views.py
+:PROPERTIES:
+:context: 
+:END:
+:LOGBOOK:
+- State "DONE"       from "TODO"       [2026-06-23 Tue 19:15]
+:END:
+
+In file src/org/commands/agenda/views.py around line 69, near the ~class AgendaViewContext:~
+
+#+begin_example
++def _build_section_query(filter_query: str, order_by: str | None) -> str:
++    """Build one agenda section filter/order query text."""
++    return build_filter_order_query_text(filter_query, order_by)
+
+#+end_example
+
+These wrappers aren't very useful, please clean this up in all the commands that now have these.
+
+**** REWORKED src/org/commands/tasks/common.py
+:PROPERTIES:
+:context: 
+:END:
+:LOGBOOK:
+- State "REWORKED"   from "TODO"       [2026-06-23 Tue 19:48]
+:END:
+
+In file src/org/commands/tasks/common.py around line 385, near the ~def resolve_headings_by_query(~
+
+#+begin_example
+     selector_query: str,
+ ) -> list[Heading]:
+     """Resolve matching headings across files from selector query."""
+-    compiled_query = compile_query_or_raise(
+-        selector_query,
+-        lambda message: typer.BadParameter(f"Invalid task selector query: {message}"),
+-    )
+-
+     logger.info("Task selector query: %s", selector_query)
+     matches_by_identity: dict[int, Heading] = {}
+     for filename in filenames:
+         document = load_document(filename)
+         logger.info("Running task selector query against file: %s", filename)
+-        results = execute_query_or_raise(
+-            compiled_query,
+-            [document],
+-            {},
+-            lambda message: typer.BadParameter(f"Task selector query failed: {message}"),
+-        )
++        try:
++            results = run_query([document], [selector_query], {})
++        except QueryParseError as exc:
++            raise typer.BadParameter(f"Invalid task selector query: {exc}") from exc
++        except QueryRuntimeError as exc:
++            raise typer.BadParameter(f"Task selector query failed: {exc}") from exc
+ 
+         for value in results:
+             if not isinstance(value, Heading):
+
+#+end_example
+
+This whole function should use the usual ETL pipeline.
+Ideally, this function should be removed and the callers should use the usual ETL pipeline.
+**** DONE src/org/commands/tasks/remove.py
+:PROPERTIES:
+:context: 
+:END:
+:LOGBOOK:
+- State "DONE"       from "TODO"       [2026-06-23 Tue 19:50]
+:END:
+
+In file src/org/commands/tasks/remove.py around line 99, near the ~def _selected_delete_roots(headings: list[Heading]) -> list[Heading]:~
+
+#+begin_example
++    matches_by_identity: dict[int, Heading] = {}
++    for value in results:
++        if not isinstance(value, Heading):
++            raise typer.BadParameter("Task selector query must return task headings")
++        matches_by_identity[id(value)] = value
++    selected_headings = list(matches_by_identity.values())
++    if not selected_headings:
+
+#+end_example
+
+Sounds like this could be extracted to be a separate function that's used by the commands that need to handle the identity.
+
+**** DONE Cache query compilation results
+:LOGBOOK:
+- State "DONE"       from "TODO"       [2026-06-23 Tue 20:09]
+:END:
+It is often the case that the same queries are recompiled multiple times. We could improve performance by caching the query compilation results and checking if the query was already compiled.
+Add the query compilation cache.
 
 *** TODO Org file repository abstraction
 :PROPERTIES:
@@ -9343,6 +9438,9 @@ org tasks query all-todos
 - Show stats widgets when query return widget objects
 - Support mixed data (title string + histogram widget, etc)
 
+*** Limit & offset applied outside of the query
+- Define variables in context
+- Apply =results[offset:offset + limit]= outside of the query.
 ** Board command improvements                                                             :UX:Board:
 *** Board plan command
 Shows long term plan including completeness of tasks that have children and gannt-style chart for execution.
diff --git a/docs/query_language.md b/docs/query_language.md
index 76c09f1..3899378 100644
--- a/docs/query_language.md
+++ b/docs/query_language.md
@@ -10,7 +10,7 @@ A query evaluates over a **stream of values**. Each stage receives a stream and
 - `.` is identity: it forwards the current stream item unchanged.
 - Many expressions can emit zero, one, or many output values per input value.
 
-At the CLI, the initial stream contains one value: the loaded roots collection.
+At the CLI, the initial stream contains the loaded Org documents.
 
 ```bash
 poetry run org tasks query '.[] | .children | length' examples/ARCHIVE_small
@@ -112,12 +112,21 @@ Some functions are no-arg and are written without parentheses:
 
 ```text
 reverse
+fold
 length
 sum
 any
 all
 ```
 
+`fold` collects the current stream into one list value. This differs from bracket fold syntax,
+which runs a subquery per input item and collects each subquery result stream separately.
+
+```text
+[1, 2, 3][] | fold
+.[] | (.children[] | .title_text | fold) | length
+```
+
 ### Tuple expression
 
 Comma builds tuples from expression outputs:
@@ -733,7 +742,7 @@ Notes:
 .[ $offset : $offset + $limit ]
 
 # find most recently modified tasks
-.[][] | sort_by(.repeats + .deadline + .closed + .scheduled | max) | .title_text
+.[] | sort_by(.repeats + .deadline + .closed + .scheduled | max) | .title_text
 
 # same time, two days from now
 now + days(2)
diff --git a/docs/tasks_query_command.md b/docs/tasks_query_command.md
index f422eb5..040ce1a 100644
--- a/docs/tasks_query_command.md
+++ b/docs/tasks_query_command.md
@@ -46,13 +46,13 @@ Shared top-level config still applies for `mapping`, `exclude`, `todo_states`, `
 1) Fetch headings from all tasks
 
 ```bash
-poetry run org tasks query '.[][] | .title_text' examples/ARCHIVE_small
+poetry run org tasks query '.[] | .title_text' examples/ARCHIVE_small
 ```
 
 2) Fetch headings of completed tasks
 
 ```bash
-poetry run org tasks query '.[][] | select(.todo in $done_states) | .title_text' \
+poetry run org tasks query '.[] | select(.todo in $done_states) | .title_text' \
   examples/ARCHIVE_small
 ```
 
@@ -67,7 +67,7 @@ Prepare a document for team onboarding to Ejabberd.
 3) Fetch a paged window of headings
 
 ```bash
-poetry run org tasks query '[ .[][] | .title_text ][ $offset : $offset + $limit ]' \
+poetry run org tasks query '[ .[] | .title_text ][ $offset : $offset + $limit ]' \
   --offset 5 \
   --limit 5 \
   examples/ARCHIVE_small
@@ -76,7 +76,7 @@ poetry run org tasks query '[ .[][] | .title_text ][ $offset : $offset + $limit
 4) Fetch headings for tasks tagged `Debugging`, then page results
 
 ```bash
-poetry run org tasks query '[ .[][] | select("Debugging" in .tags) | .title_text ][ $offset : $offset + $limit ]' \
+poetry run org tasks query '[ .[] | select("Debugging" in .tags) | .title_text ][ $offset : $offset + $limit ]' \
   --offset 2 \
   --limit 8 \
   examples/ARCHIVE_small
@@ -85,7 +85,16 @@ poetry run org tasks query '[ .[][] | select("Debugging" in .tags) | .title_text
 5) Fetch tasks that are not completed yet
 
 ```bash
-poetry run org tasks query '[ .[][] | select(not(.todo in $done_states)) ][ $offset : $offset + $limit ]' \
+poetry run org tasks query '[ .[] | select(not(.todo in $done_states)) ][ $offset : $offset + $limit ]' \
+  --offset 0 \
+  --limit 5 \
+  examples/ARCHIVE_small
+```
+
+If you want to page across the whole current stream at that point, use `fold` first:
+
+```bash
+poetry run org tasks query '.[] | select(not(.todo in $done_states)) | fold | .[$offset:$offset + $limit]' \
   --offset 0 \
   --limit 5 \
   examples/ARCHIVE_small
@@ -119,7 +128,7 @@ SCHEDULED: <2023-10-19 czw>
 6) Export markdown with a custom Pygments theme
 
 ```bash
-poetry run org tasks query '.[][] | .title_text' \
+poetry run org tasks query '.[] | .title_text' \
   --out gfm \
   --out-theme monokai \
   examples/ARCHIVE_small
@@ -128,5 +137,5 @@ poetry run org tasks query '.[][] | .title_text' \
 7) Export PDF by piping raw binary output to a file
 
 ```bash
-poetry run org tasks query '.[][]' --out pdf examples/ARCHIVE_small > report.pdf
+poetry run org tasks query '.[]' --out pdf examples/ARCHIVE_small > report.pdf
 ```
diff --git a/src/org/commands/agenda/ui.py b/src/org/commands/agenda/ui.py
index 454a108..1e409ba 100644
--- a/src/org/commands/agenda/ui.py
+++ b/src/org/commands/agenda/ui.py
@@ -13,7 +13,8 @@
 from rich.text import Text
 
 from org.logic.time import local_now, parse_date_argument, shift_datetimes_by_unit
-from org.query.runner import execute_query_or_raise
+from org.query.engine.errors import QueryParseError, QueryRuntimeError
+from org.query.runner import run_query
 from org.tui.bits import (
     heading_title_to_text,
     task_priority_to_text,
@@ -787,15 +788,16 @@ def _apply_section_query(
     view_name: str,
 ) -> list[Heading]:
     """Apply section query against nodes with $date in context, returning matched nodes."""
-    results = execute_query_or_raise(
-        spec.query,
-        nodes,
-        {"date": day},
-        lambda message: typer.BadParameter(
-            "Agenda filter/order-by query failed "
-            f"(section={spec.name}, view={view_name}): {message}",
-        ),
-    )
+    try:
+        results = run_query(nodes, [spec.query_text], {"date": day})
+    except QueryParseError as exc:
+        raise typer.BadParameter(
+            f"Invalid agenda filter/order-by (view={view_name}, section={spec.name}): {exc}",
+        ) from exc
+    except QueryRuntimeError as exc:
+        raise typer.BadParameter(
+            f"Agenda filter/order-by query failed (section={spec.name}, view={view_name}): {exc}",
+        ) from exc
     return [cast("Heading", node) for node in results]
 
 
diff --git a/src/org/commands/agenda/views.py b/src/org/commands/agenda/views.py
index e02bc9b..173d481 100644
--- a/src/org/commands/agenda/views.py
+++ b/src/org/commands/agenda/views.py
@@ -8,21 +8,20 @@
 import typer
 
 import org.config.app
-from org.query.runner import compile_filter_order_query
+from org.query.engine.errors import QueryParseError, QueryRuntimeError
+from org.query.runner import build_filter_order_query_text, run_query
 
 
 if TYPE_CHECKING:
-    from org.query.engine.compiler import CompiledQuery
-
     from .command import AgendaArgs
 
 
 @dataclass(frozen=True)
 class AgendaSectionSpec:
-    """Compiled agenda section specification."""
+    """Agenda section query specification."""
 
     name: str
-    query: CompiledQuery
+    query_text: str
     style: str
     timeline: bool
 
@@ -35,11 +34,6 @@ class AgendaViewContext:
     name: str
 
 
-def _compile_section_query(filter_query: str, order_by: str | None) -> CompiledQuery:
-    """Compile one agenda section filter/order query text."""
-    return compile_filter_order_query(filter_query, order_by)
-
-
 def _fallback_agenda_view() -> org.config.app.AgendaViewConfig:
     """Return built-in fallback agenda view definition."""
     return org.config.app.AgendaViewConfig(
@@ -61,15 +55,16 @@ def _compile_view_section_specs(view: org.config.app.AgendaViewConfig) -> list[A
     section_specs: list[AgendaSectionSpec] = []
     for section in view.sections:
         try:
-            query = _compile_section_query(section.filter, section.order_by)
-        except Exception as err:
+            query_text = build_filter_order_query_text(section.filter, section.order_by)
+            run_query([], [query_text], {})
+        except (QueryParseError, QueryRuntimeError) as err:
             raise typer.BadParameter(
                 f"Invalid agenda filter/order-by (view={view.name}, section={section.name}): {err}",
             ) from err
         section_specs.append(
             AgendaSectionSpec(
                 name=section.name,
-                query=query,
+                query_text=query_text,
                 style=section.style,
                 timeline=section.timeline,
             ),
diff --git a/src/org/commands/board/actions.py b/src/org/commands/board/actions.py
index edf9ef1..f2dffd2 100644
--- a/src/org/commands/board/actions.py
+++ b/src/org/commands/board/actions.py
@@ -24,7 +24,8 @@
 )
 from org.logic.time import advance_timestamp_by_repeater, local_now
 from org.pipeline.load import load_and_process_data
-from org.query.runner import compile_filter_order_query, execute_query_or_raise
+from org.query.engine.errors import QueryParseError, QueryRuntimeError
+from org.query.runner import build_filter_order_query_text, run_query
 
 
 if TYPE_CHECKING:
@@ -33,7 +34,6 @@
     from org_parser.document import Document, Heading
 
     from org.config.app import AppConfig, BoardViewConfig
-    from org.query.engine.compiler import CompiledQuery
 
     from .command import BoardArgs
 
@@ -70,7 +70,7 @@ class BoardColumnSpec:
 
     name: str
     view_name: str
-    query: CompiledQuery
+    query_text: str
 
 
 @dataclass
@@ -148,14 +148,6 @@ def resolved_states(
     return todo_states, done_states
 
 
-def _compile_column_filter_query(
-    filter_query: str,
-    order_by: str | None,
-) -> CompiledQuery:
-    """Compile one board column filter/order query text."""
-    return compile_filter_order_query(filter_query, order_by)
-
-
 def _fallback_view_config() -> org.config.app.BoardViewConfig:
     """Return built-in fallback board filter view definition."""
     return org.config.app.BoardViewConfig(
@@ -176,8 +168,9 @@ def compile_view_column_specs(view: org.config.app.BoardViewConfig) -> list[Boar
     column_specs: list[BoardColumnSpec] = []
     for column in view.columns:
         try:
-            query = _compile_column_filter_query(column.filter, column.order_by)
-        except Exception as err:
+            query_text = build_filter_order_query_text(column.filter, column.order_by)
+            run_query([], [query_text], {})
+        except (QueryParseError, QueryRuntimeError) as err:
             raise typer.BadParameter(
                 f"Invalid board filter/order-by (view={view.name}, column={column.name}): {err}",
             ) from err
@@ -185,7 +178,7 @@ def compile_view_column_specs(view: org.config.app.BoardViewConfig) -> list[Boar
             BoardColumnSpec(
                 name=column.name,
                 view_name=view.name,
-                query=query,
+                query_text=query_text,
             ),
         )
     return column_specs
@@ -228,12 +221,14 @@ def build_selector_board_columns(
     """Evaluate filter specs against processed task stream."""
     columns: list[BoardColumn] = []
     for spec in column_specs:
-        results = execute_query_or_raise(
-            spec.query,
-            nodes,
-            {},
-            _board_query_error_builder(spec.view_name, spec.name),
-        )
+        try:
+            results = run_query(nodes, [spec.query_text], {})
+        except QueryParseError as exc:
+            raise typer.BadParameter(
+                f"Invalid board filter/order-by (view={spec.view_name}, column={spec.name}): {exc}",
+            ) from exc
+        except QueryRuntimeError as exc:
+            raise _board_query_error_builder(spec.view_name, spec.name)(str(exc)) from exc
 
         column_nodes = [cast("Heading", node) for node in results]
         columns.append(BoardColumn(title=spec.name, nodes=column_nodes))
diff --git a/src/org/commands/tasks/archive.py b/src/org/commands/tasks/archive.py
index 3290956..8fb75dc 100644
--- a/src/org/commands/tasks/archive.py
+++ b/src/org/commands/tasks/archive.py
@@ -11,12 +11,14 @@
 import org.config.app
 import org.logging
 from org.commands.tasks.common import (
-    resolve_headings_by_query,
     resolve_task_selector_query,
     save_document,
+    selected_heading_results,
 )
 from org.logic.archive import archive_heading_subtree, archive_result_documents_to_save
-from org.pipeline.load import resolve_input_paths
+from org.pipeline.load import load_documents, resolve_input_paths, resolve_loaded_state_context
+from org.query.engine.errors import QueryParseError, QueryRuntimeError
+from org.query.runner import run_query
 
 
 logger = logging.getLogger("org")
@@ -52,10 +54,30 @@ def _selected_archive_roots(headings: list[Heading]) -> list[Heading]:
 
 def run_tasks_archive(args: ArchiveArgs, config: org.config.app.AppConfig) -> None:
     """Run the tasks archive command."""
-    del config
     filenames = resolve_input_paths(args.files)
     selector_query = resolve_task_selector_query(args.query_title, args.query_id, args.query)
-    selected_headings = resolve_headings_by_query(filenames, selector_query)
+    documents = load_documents(filenames)
+    todo_states, done_states = resolve_loaded_state_context(
+        documents,
+        config.todo_states,
+        config.done_states,
+    )
+
+    logger.info("Task selector query: %s", selector_query)
+    try:
+        results = run_query(
+            documents,
+            [selector_query],
+            {"todo_states": todo_states, "done_states": done_states},
+        )
+    except QueryParseError as exc:
+        raise typer.BadParameter(f"Invalid task selector query: {exc}") from exc
+    except QueryRuntimeError as exc:
+        raise typer.BadParameter(f"Task selector query failed: {exc}") from exc
+
+    selected_headings = selected_heading_results(results)
+    if not selected_headings:
+        raise typer.BadParameter("No task matches the provided selector")
     archive_roots = _selected_archive_roots(selected_headings)
     destination_cache: dict[str, Document] = {}
 
diff --git a/src/org/commands/tasks/capture/domain.py b/src/org/commands/tasks/capture/domain.py
index ab08825..5776419 100644
--- a/src/org/commands/tasks/capture/domain.py
+++ b/src/org/commands/tasks/capture/domain.py
@@ -13,9 +13,15 @@
 from org_parser.document import Document, Heading
 from rich.syntax import Syntax
 
-from org.commands.tasks.common import load_document, resolve_parent_heading, save_document
+from org.commands.tasks.common import (
+    load_document,
+    resolve_parent_heading,
+    save_document,
+    selected_heading_results,
+)
 from org.pipeline.format import DEFAULT_OUTPUT_THEME
-from org.query.runner import compile_query_or_raise, execute_query_or_raise
+from org.query.engine.errors import QueryParseError, QueryRuntimeError
+from org.query.runner import run_query
 from org.tui.help import InteractiveHelpEntry
 
 
@@ -297,25 +303,17 @@ def _resolve_parent_from_selector(document: Document, parent_selector: str) -> H
         raise typer.BadParameter("Capture template parent selector cannot be empty")
 
     query_text = f".[] | select({normalized_selector})"
+    try:
+        results = run_query([document], [query_text], {})
+    except QueryParseError as exc:
+        raise typer.BadParameter(f"Invalid parent selector: {exc}") from exc
+    except QueryRuntimeError as exc:
+        raise typer.BadParameter(f"Parent selector failed: {exc}") from exc
 
-    compiled_query = compile_query_or_raise(
-        query_text,
-        lambda message: typer.BadParameter(f"Invalid parent selector: {message}"),
-    )
-    results = execute_query_or_raise(
-        compiled_query,
-        [document],
-        {},
-        lambda message: typer.BadParameter(f"Parent selector failed: {message}"),
-    )
-
-    matches_by_identity: dict[int, Heading] = {}
-    for result in results:
-        if not isinstance(result, Heading):
-            raise typer.BadParameter("Parent selector must match task headings")
-        matches_by_identity[id(result)] = result
-
-    matches = list(matches_by_identity.values())
+    try:
+        matches = selected_heading_results(results)
+    except typer.BadParameter as exc:
+        raise typer.BadParameter("Parent selector must match task headings") from exc
     if not matches:
         raise typer.BadParameter("Parent selector did not match any heading")
     if len(matches) > 1:
diff --git a/src/org/commands/tasks/common.py b/src/org/commands/tasks/common.py
index 603b4f4..f5a2f08 100644
--- a/src/org/commands/tasks/common.py
+++ b/src/org/commands/tasks/common.py
@@ -13,8 +13,6 @@
 from org_parser.text import CompletionCounter
 from org_parser.time import Timestamp
 
-from org.query.runner import compile_query_or_raise, execute_query_or_raise
-
 
 if TYPE_CHECKING:
     from org_parser.document import Document
@@ -174,6 +172,16 @@ def resolve_parent_heading(document: Document, parent_value: str) -> Heading:
     raise typer.BadParameter(f"--parent '{selector}' was not found")
 
 
+def selected_heading_results(results: list[object]) -> list[Heading]:
+    """Validate query results as headings and deduplicate by object identity."""
+    matches_by_identity: dict[int, Heading] = {}
+    for value in results:
+        if not isinstance(value, Heading):
+            raise typer.BadParameter("Task selector query must return task headings")
+        matches_by_identity[id(value)] = value
+    return list(matches_by_identity.values())
+
+
 def parse_tags_csv(value: str) -> list[str]:
     """Parse comma-separated tag option into tag list."""
     normalized = value.strip()
@@ -349,38 +357,3 @@ def parse_properties_json(value: str) -> dict[str, str]:
             raise typer.BadParameter("--properties values must be strings")
         parsed[key] = property_value
     return parsed
-
-
-def resolve_headings_by_query(
-    filenames: list[str],
-    selector_query: str,
-) -> list[Heading]:
-    """Resolve matching headings across files from selector query."""
-    compiled_query = compile_query_or_raise(
-        selector_query,
-        lambda message: typer.BadParameter(f"Invalid task selector query: {message}"),
-    )
-
-    logger.info("Task selector query: %s", selector_query)
-    matches_by_identity: dict[int, Heading] = {}
-    for filename in filenames:
-        document = load_document(filename)
-        logger.info("Running task selector query against file: %s", filename)
-        results = execute_query_or_raise(
-            compiled_query,
-            [document],
-            {},
-            lambda message: typer.BadParameter(f"Task selector query failed: {message}"),
-        )
-
-        for value in results:
-            if not isinstance(value, Heading):
-                raise typer.BadParameter(
-                    "Task selector query must return task headings",
-                )
-            matches_by_identity[id(value)] = value
-
-    matches = list(matches_by_identity.values())
-    if not matches:
-        raise typer.BadParameter("No task matches the provided selector")
-    return matches
diff --git a/src/org/commands/tasks/edit.py b/src/org/commands/tasks/edit.py
index 1b45fd0..77c1b9a 100644
--- a/src/org/commands/tasks/edit.py
+++ b/src/org/commands/tasks/edit.py
@@ -10,11 +10,13 @@
 import org.config.app
 import org.logging
 from org.commands.tasks.common import (
-    resolve_headings_by_query,
     resolve_task_selector_query,
+    selected_heading_results,
 )
 from org.logic.edit import edit_heading_subtree_in_external_editor
-from org.pipeline.load import resolve_input_paths
+from org.pipeline.load import load_documents, resolve_input_paths, resolve_loaded_state_context
+from org.query.engine.errors import QueryParseError, QueryRuntimeError
+from org.query.runner import run_query
 
 
 logger = logging.getLogger("org")
@@ -33,10 +35,30 @@ class EditArgs:
 
 def run_tasks_edit(args: EditArgs, config: org.config.app.AppConfig) -> None:
     """Run the tasks edit command."""
-    del config
     filenames = resolve_input_paths(args.files)
     selector_query = resolve_task_selector_query(args.query_title, args.query_id, args.query)
-    selected_headings = resolve_headings_by_query(filenames, selector_query)
+    documents = load_documents(filenames)
+    todo_states, done_states = resolve_loaded_state_context(
+        documents,
+        config.todo_states,
+        config.done_states,
+    )
+
+    logger.info("Task selector query: %s", selector_query)
+    try:
+        results = run_query(
+            documents,
+            [selector_query],
+            {"todo_states": todo_states, "done_states": done_states},
+        )
+    except QueryParseError as exc:
+        raise typer.BadParameter(f"Invalid task selector query: {exc}") from exc
+    except QueryRuntimeError as exc:
+        raise typer.BadParameter(f"Task selector query failed: {exc}") from exc
+
+    selected_headings = selected_heading_results(results)
+    if not selected_headings:
+        raise typer.BadParameter("No task matches the provided selector")
 
     if len(selected_headings) > 1:
         raise typer.BadParameter("tasks edit requires a selector that matches exactly one task")
diff --git a/src/org/commands/tasks/find.py b/src/org/commands/tasks/find.py
index 3dc861e..30c4dc3 100644
--- a/src/org/commands/tasks/find.py
+++ b/src/org/commands/tasks/find.py
@@ -22,7 +22,8 @@
     print_prepared_output,
 )
 from org.pipeline.load import load_root_data
-from org.query.runner import compile_query_or_raise, execute_query_or_raise
+from org.query.engine.errors import QueryParseError, QueryRuntimeError
+from org.query.runner import run_query
 from org.tui.bits import build_console, processing_status, setup_output
 
 
@@ -140,16 +141,16 @@ def run_tasks_find(args: TasksFindArgs, config: org.config.app.AppConfig) -> Non
 
     with processing_status(console, color_enabled):
         query_text = _build_find_query(args)
-        compiled_query = compile_query_or_raise(query_text, click.UsageError)
 
         roots, todo_states, done_states = load_root_data(args)
-        nodes = [node for root in roots for node in list(root)]
-        results = execute_query_or_raise(
-            compiled_query,
-            [nodes],
-            {"todo_states": todo_states, "done_states": done_states},
-            click.UsageError,
-        )
+        try:
+            results = run_query(
+                roots,
+                [query_text],
+                {"todo_states": todo_states, "done_states": done_states},
+            )
+        except (QueryParseError, QueryRuntimeError) as exc:
+            raise click.UsageError(str(exc)) from exc
 
         matched_nodes = [value for value in results if isinstance(value, Heading)]
         output_nodes = _nodes_with_context(matched_nodes, args.include_context)
diff --git a/src/org/commands/tasks/query.py b/src/org/commands/tasks/query.py
index c11d8c8..de0181b 100644
--- a/src/org/commands/tasks/query.py
+++ b/src/org/commands/tasks/query.py
@@ -32,7 +32,8 @@
     print_prepared_output,
 )
 from org.pipeline.load import load_root_data
-from org.query.runner import compile_query_or_raise, execute_query_or_raise
+from org.query.engine.errors import QueryParseError, QueryRuntimeError
+from org.query.runner import run_query
 from org.tui.bits import build_console, processing_status, setup_output
 
 
@@ -234,8 +235,6 @@ def run_tasks_query(args: TasksQueryArgs, config: org.config.app.AppConfig) -> N
         raise click.UsageError(str(exc)) from exc
 
     with processing_status(console, color_enabled):
-        compiled_query = compile_query_or_raise(args.query, click.UsageError)
-
         roots, todo_states, done_states = load_root_data(args)
 
         context = {
@@ -244,7 +243,10 @@ def run_tasks_query(args: TasksQueryArgs, config: org.config.app.AppConfig) -> N
             "todo_states": todo_states,
             "done_states": done_states,
         }
-        results = execute_query_or_raise(compiled_query, [roots], context, click.UsageError)
+        try:
+            results = run_query(roots, [args.query], context)
+        except (QueryParseError, QueryRuntimeError) as exc:
+            raise click.UsageError(str(exc)) from exc
 
         first_result = results[0] if results else None
         if len(results) == 1 and isinstance(first_result, list | tuple | set):
diff --git a/src/org/commands/tasks/remove.py b/src/org/commands/tasks/remove.py
index 19454bf..aa494a4 100644
--- a/src/org/commands/tasks/remove.py
+++ b/src/org/commands/tasks/remove.py
@@ -13,11 +13,13 @@
 import org.config.app
 import org.logging
 from org.commands.tasks.common import (
-    resolve_headings_by_query,
     resolve_task_selector_query,
     save_document,
+    selected_heading_results,
 )
-from org.pipeline.load import resolve_input_paths
+from org.pipeline.load import load_documents, resolve_input_paths, resolve_loaded_state_context
+from org.query.engine.errors import QueryParseError, QueryRuntimeError
+from org.query.runner import run_query
 from org.tui.color import should_use_color
 
 
@@ -64,11 +66,30 @@ def _selected_delete_roots(headings: list[Heading]) -> list[Heading]:
 
 def run_tasks_remove(args: RemoveArgs, config: org.config.app.AppConfig) -> None:
     """Run the tasks remove command."""
-    del config
     filenames = resolve_input_paths(args.files)
     selector_query = resolve_task_selector_query(args.query_title, args.query_id, args.query)
-
-    selected_headings = resolve_headings_by_query(filenames, selector_query)
+    documents = load_documents(filenames)
+    todo_states, done_states = resolve_loaded_state_context(
+        documents,
+        config.todo_states,
+        config.done_states,
+    )
+
+    logger.info("Task selector query: %s", selector_query)
+    try:
+        results = run_query(
+            documents,
+            [selector_query],
+            {"todo_states": todo_states, "done_states": done_states},
+        )
+    except QueryParseError as exc:
+        raise typer.BadParameter(f"Invalid task selector query: {exc}") from exc
+    except QueryRuntimeError as exc:
+        raise typer.BadParameter(f"Task selector query failed: {exc}") from exc
+
+    selected_headings = selected_heading_results(results)
+    if not selected_headings:
+        raise typer.BadParameter("No task matches the provided selector")
     selected_count = len(selected_headings)
     logger.info("Selected %s tasks for delete", selected_count)
 
diff --git a/src/org/commands/tasks/update.py b/src/org/commands/tasks/update.py
index 94c9c8a..07db441 100644
--- a/src/org/commands/tasks/update.py
+++ b/src/org/commands/tasks/update.py
@@ -25,12 +25,14 @@
     parse_planning_timestamp,
     parse_properties_json,
     parse_tags_csv,
-    resolve_headings_by_query,
     resolve_parent_heading,
     resolve_task_selector_query,
     save_document,
+    selected_heading_results,
 )
-from org.pipeline.load import resolve_input_paths
+from org.pipeline.load import load_documents, resolve_input_paths, resolve_loaded_state_context
+from org.query.engine.errors import QueryParseError, QueryRuntimeError
+from org.query.runner import run_query
 from org.tui.color import should_use_color
 
 
@@ -398,14 +400,43 @@ def _apply_field_updates(args: UpdateArgs, heading: Heading) -> None:
         heading.body = args.body
 
 
+def _select_headings_for_update(
+    filenames: list[str],
+    selector_query: str,
+    config: org.config.app.AppConfig,
+) -> list[Heading]:
+    """Load documents and resolve selected headings for update."""
+    documents = load_documents(filenames)
+    todo_states, done_states = resolve_loaded_state_context(
+        documents,
+        config.todo_states,
+        config.done_states,
+    )
+
+    logger.info("Task selector query: %s", selector_query)
+    try:
+        results = run_query(
+            documents,
+            [selector_query],
+            {"todo_states": todo_states, "done_states": done_states},
+        )
+    except QueryParseError as exc:
+        raise typer.BadParameter(f"Invalid task selector query: {exc}") from exc
+    except QueryRuntimeError as exc:
+        raise typer.BadParameter(f"Task selector query failed: {exc}") from exc
+
+    headings = selected_heading_results(results)
+    if not headings:
+        raise typer.BadParameter("No task matches the provided selector")
+    return headings
+
+
 def run_tasks_update(args: UpdateArgs, config: org.config.app.AppConfig) -> None:
     """Run the tasks update command."""
-    del config
     selector_query = resolve_task_selector_query(args.query_title, args.query_id, args.query)
     _validate_update_option_conflicts(args)
     filenames = resolve_input_paths(args.files)
-
-    headings = resolve_headings_by_query(filenames, selector_query)
+    headings = _select_headings_for_update(filenames, selector_query, config)
     selected_count = len(headings)
     logger.info("Selected %s tasks for update", selected_count)
 
diff --git a/src/org/pipeline/load.py b/src/org/pipeline/load.py
index cb1f99a..7ad5599 100644
--- a/src/org/pipeline/load.py
+++ b/src/org/pipeline/load.py
@@ -8,6 +8,7 @@
 from typing import TYPE_CHECKING, Protocol, cast
 
 import typer
+from org_parser import load as load_org_document
 from org_parser import loads
 from org_parser.document import Heading
 
@@ -19,12 +20,8 @@
     validate_custom_switches,
 )
 from org.logic.validation import validate_and_parse_keys, validate_global_arguments
-from org.query.engine.errors import QueryParseError
-from org.query.runner import (
-    build_query_from_stages,
-    execute_query_or_raise,
-    flatten_query_results,
-)
+from org.query.engine.errors import QueryParseError, QueryRuntimeError
+from org.query.runner import run_query
 
 
 if TYPE_CHECKING:
@@ -99,6 +96,37 @@ def load_root_nodes(
     return roots, all_todo_states, all_done_states
 
 
+def load_documents(
+    filenames: list[str],
+) -> list[Document]:
+    """Load org-mode files and return parsed documents."""
+    documents: list[Document] = []
+    for filename in filenames:
+        try:
+            documents.append(load_org_document(filename))
+        except FileNotFoundError as err:
+            raise typer.BadParameter(f"File '{filename}' not found") from err
+        except PermissionError as err:
+            raise typer.BadParameter(f"Permission denied for '{filename}'") from err
+        except ValueError as err:
+            raise typer.BadParameter(f"Unable to parse '{filename}': {err}") from err
+    return documents
+
+
+def resolve_loaded_state_context(
+    documents: list[Document],
+    todo_states: list[str],
+    done_states: list[str],
+) -> tuple[list[str], list[str]]:
+    """Merge discovered todo and done states from loaded documents."""
+    all_todo_states = list(todo_states)
+    all_done_states = list(done_states)
+    for document in documents:
+        all_todo_states = _merge_state_order(all_todo_states, list(document.todo_states))
+        all_done_states = _merge_state_order(all_done_states, list(document.done_states))
+    return all_todo_states, all_done_states
+
+
 def resolve_input_paths(inputs: list[str] | None) -> list[str]:  # noqa: C901
     """Resolve CLI inputs into a list of org files to process."""
     resolved_files: list[str] = []
@@ -205,16 +233,9 @@ def load_and_process_data(
         todo_states,
         done_states,
     )
-    nodes = [node for root in roots for node in list(root)]
 
     include_slice = include_ordering and hasattr(args, "offset") and hasattr(args, "max_results")
-    try:
-        query = build_query_from_stages(
-            build_pipeline_stages(config, args, sys.argv, include_ordering),
-            include_slice,
-        )
-    except QueryParseError as exc:
-        raise typer.BadParameter(str(exc)) from exc
+    stages = [".[]", *build_pipeline_stages(config, args, sys.argv, include_ordering)]
 
     context_vars: dict[str, object] = {
         "todo_states": todo_states,
@@ -228,7 +249,15 @@ def load_and_process_data(
 
     logger.info("Query context: %s", context_vars)
 
-    results = execute_query_or_raise(query, [nodes], context_vars, typer.BadParameter)
+    try:
+        results = run_query(roots, stages, context_vars)
+    except (QueryParseError, QueryRuntimeError) as exc:
+        raise typer.BadParameter(str(exc)) from exc
 
-    flattened = flatten_query_results(results)
-    return [value for value in flattened if isinstance(value, Heading)], todo_states, done_states
+    heading_results = [value for value in results if isinstance(value, Heading)]
+    if include_slice:
+        sliced_args = cast("SlicedDataLoadArgs", args)
+        heading_results = heading_results[
+            sliced_args.offset : sliced_args.offset + sliced_args.max_results
+        ]
+    return heading_results, todo_states, done_states
diff --git a/src/org/query/engine/interpreter.py b/src/org/query/engine/interpreter.py
index 0fda1ac..63164c3 100644
--- a/src/org/query/engine/interpreter.py
+++ b/src/org/query/engine/interpreter.py
@@ -612,17 +612,21 @@ def _evaluate_fold(expr: Fold, stream: Stream, context: EvalContext) -> Stream:
 
     output = _stream()
     for item in stream:
-        folded_values = evaluate_expr(expr.expr, _stream([item]), context)
-        folded: list[object] = []
-        for value in folded_values:
-            if isinstance(value, tuple):
-                folded.extend(list(value))
-                continue
-            folded.append(value)
-        output.append(folded)
+        output.append(_fold_stream_values(evaluate_expr(expr.expr, _stream([item]), context)))
     return output
 
 
+def _fold_stream_values(values: Stream) -> list[object]:
+    """Collect one stream into a list value using fold semantics."""
+    folded: list[object] = []
+    for value in values:
+        if isinstance(value, tuple):
+            folded.extend(list(value))
+            continue
+        folded.append(value)
+    return folded
+
+
 def _resolve_field(value: object, field: str) -> object:
     """Resolve attribute-like field access with None fallback."""
     if value is None:
@@ -1119,6 +1123,7 @@ def _evaluate_function(expr: FunctionCall, stream: Stream, context: EvalContext)
         "analyze": _func_analyze,
         "all": _func_all,
         "any": _func_any,
+        "fold": _func_fold,
         "now": _func_now,
         "reverse": _func_reverse,
         "unique": _func_unique,
@@ -1378,6 +1383,11 @@ def _func_now(stream: Stream) -> Stream:
     return _stream([_current_datetime() for _item in stream])
 
 
+def _func_fold(stream: Stream) -> Stream:
+    """Collect the current stream into one list value."""
+    return _stream([_fold_stream_values(stream)])
+
+
 def _func_type(stream: Stream) -> Stream:
     """Return type names for stream values."""
     return _stream([_type_name(value) for value in stream])
diff --git a/src/org/query/engine/parser.py b/src/org/query/engine/parser.py
index 68a2d7d..b70a2a1 100644
--- a/src/org/query/engine/parser.py
+++ b/src/org/query/engine/parser.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import ast
+from functools import lru_cache
 from typing import TYPE_CHECKING, Literal, cast
 
 from parsy import ParseError, Parser, eof, forward_declaration, generate, regex, seq, string
@@ -48,6 +49,7 @@
     "datetime",
     "days",
     "float",
+    "fold",
     "hours",
     "int",
     "reverse",
@@ -649,6 +651,7 @@ def parser() -> Generator[Parser, object, Expr]:
 QUERY_PARSER = _make_parser()
 
 
+@lru_cache(maxsize=256)
 def parse_query(query: str) -> Expr:
     """Parse query text into an AST expression."""
     try:
diff --git a/src/org/query/runner.py b/src/org/query/runner.py
index 1fb21cc..218eccc 100644
--- a/src/org/query/runner.py
+++ b/src/org/query/runner.py
@@ -1,20 +1,23 @@
-"""Query text assembly, compilation, and execution helpers."""
+"""Query text assembly and execution helpers."""
 
 from __future__ import annotations
 
 import json
 import logging
-from collections.abc import Callable, Sequence
-from typing import cast
+from typing import TYPE_CHECKING
 
-from org.query.engine.compiler import CompiledQuery, compile_query_text
-from org.query.engine.errors import QueryParseError, QueryRuntimeError
+from org.query.engine.compiler import compile_query_text
 from org.query.engine.interpreter import EvalContext, Stream
 
 
-logger = logging.getLogger("org")
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+    from org_parser import Document
+    from org_parser.document import Heading
 
-ErrorBuilder = Callable[[str], Exception]
+
+logger = logging.getLogger("org")
 
 
 def build_custom_stage(query: str, arg_value: object) -> str:
@@ -33,75 +36,26 @@ def _query_literal(value: object) -> str:
     return json.dumps(value)
 
 
-def build_query_text_from_stages(stages: Sequence[str], include_slice: bool) -> str:
+def build_query_text_from_stages(stages: Sequence[str]) -> str:
     """Build query text from an ordered list of pipeline stages."""
-    pipeline_body = " | ".join(stages)
-    base_query = f"[ .[] | {pipeline_body} ]" if pipeline_body else "[ .[] ]"
-    return f"{base_query}[$offset:($offset + $limit)]" if include_slice else base_query
-
-
-def build_query_from_stages(stages: Sequence[str], include_slice: bool) -> CompiledQuery:
-    """Compile query from an ordered list of pipeline stages."""
-    query_text = build_query_text_from_stages(stages, include_slice)
-    logger.info("Query: %s", query_text)
-    return compile_query_text(query_text)
-
-
-def compile_query_or_raise(query_text: str, error_builder: ErrorBuilder) -> CompiledQuery:
-    """Compile query text and convert parse failures to caller-specific errors."""
-    try:
-        return compile_query_text(query_text)
-    except QueryParseError as exc:
-        raise error_builder(str(exc)) from exc
+    return " | ".join(stages) if stages else "."
 
 
-def compile_filter_order_query(filter_query: str, order_by: str | None) -> CompiledQuery:
-    """Compile a `select(...)` query with optional sort stage."""
+def build_filter_order_query_text(filter_query: str, order_by: str | None) -> str:
+    """Build a `select(...)` query with optional sort stage."""
     base_query = f"select({filter_query})"
     if order_by is None:
-        return compile_query_text(base_query)
-    return compile_query_text(f"{base_query} | sort_by({order_by})")
+        return base_query
+    return f"{base_query} | sort_by({order_by})"
 
 
-def execute_query(
-    compiled_query: CompiledQuery,
-    stream_values: Sequence[object],
+def run_query(
+    inputs: Sequence[Document | Heading],
+    stages: Sequence[str],
     context_vars: dict[str, object],
 ) -> list[object]:
-    """Execute a compiled query over stream input values."""
-    return list(compiled_query(Stream(stream_values), EvalContext(context_vars)))
-
-
-def execute_query_or_raise(
-    compiled_query: CompiledQuery,
-    stream_values: Sequence[object],
-    context_vars: dict[str, object],
-    error_builder: ErrorBuilder,
-) -> list[object]:
-    """Execute query and convert runtime failures to caller-specific errors."""
-    try:
-        return execute_query(compiled_query, stream_values, context_vars)
-    except QueryRuntimeError as exc:
-        raise error_builder(str(exc)) from exc
-
-
-def run_query_text_or_raise(
-    query_text: str,
-    stream_values: Sequence[object],
-    context_vars: dict[str, object],
-    error_builder: ErrorBuilder,
-) -> list[object]:
-    """Compile and execute query text with caller-specific error conversion."""
-    return execute_query_or_raise(
-        compile_query_or_raise(query_text, error_builder),
-        stream_values,
-        context_vars,
-        error_builder,
-    )
-
-
-def flatten_query_results(results: list[object]) -> list[object]:
-    """Flatten common single-list query result shape."""
-    if len(results) == 1 and isinstance(results[0], list):
-        return cast("list[object]", results[0])
-    return list(results)
+    """Compile and execute query stages over the provided input stream."""
+    query_text = build_query_text_from_stages(stages)
+    logger.info("Query: %s", query_text)
+    compiled_query = compile_query_text(query_text)
+    return list(compiled_query(Stream(inputs), EvalContext(context_vars)))
diff --git a/tests/cli/test_cli_runner.py b/tests/cli/test_cli_runner.py
index 91fadb0..5e39769 100644
--- a/tests/cli/test_cli_runner.py
+++ b/tests/cli/test_cli_runner.py
@@ -779,7 +779,7 @@ def test_cli_runner_allows_missing_files_when_some_exist() -> None:
     runner = CliRunner()
     fixture_path = str((FIXTURES_DIR / "multiple_tags.org").resolve())
     missing_path = str((FIXTURES_DIR / "missing.org").resolve())
-    command_args = ["tasks", "query", ".[] | .children | length", missing_path, fixture_path]
+    command_args = ["tasks", "query", ".children | length", missing_path, fixture_path]
 
     result = runner.invoke(app, command_args)
 
diff --git a/tests/cli_common/test_filter_construction.py b/tests/cli_common/test_filter_construction.py
index 36e9e78..84065eb 100644
--- a/tests/cli_common/test_filter_construction.py
+++ b/tests/cli_common/test_filter_construction.py
@@ -17,7 +17,6 @@ def _build_query_text(
     argv: list[str],
     *,
     include_ordering: bool,
-    include_slice: bool,
     config: AppConfig | None = None,
 ) -> str:
     from org.config.cli import build_pipeline_stages
@@ -29,19 +28,18 @@ def _build_query_text(
         argv,
         include_ordering,
     )
-    return build_query_text_from_stages(stages, include_slice)
+    return build_query_text_from_stages(stages)
 
 
-def _build_query(
+def _run_query(
     args: FilterArgsStub,
     argv: list[str],
     *,
     include_ordering: bool,
-    include_slice: bool,
     config: AppConfig | None = None,
 ) -> object:
     from org.config.cli import build_pipeline_stages
-    from org.query.runner import build_query_from_stages
+    from org.query.runner import run_query
 
     stages = build_pipeline_stages(
         AppConfig(config_path=".org-cli.yaml") if config is None else config,
@@ -49,7 +47,7 @@ def _build_query(
         argv,
         include_ordering,
     )
-    return build_query_from_stages(stages, include_slice)
+    return run_query([], stages, {})
 
 
 def _build_config(
@@ -115,13 +113,13 @@ def test_build_query_text_filters_only() -> None:
     args = make_args(filter_tags=["simple"])
     argv = ["org", "stats", "all", "--filter-tag", "simple", "file.org"]
 
-    query = _build_query_text(args, argv, include_ordering=False, include_slice=False)
+    query = _build_query_text(args, argv, include_ordering=False)
 
-    assert query == '[ .[] | select(.tags[] matches "simple") ]'
+    assert query == 'select(.tags[] matches "simple")'
 
 
-def test_build_query_text_with_ordering_and_slice() -> None:
-    """Query text should append order stages then slice."""
+def test_build_query_text_with_ordering() -> None:
+    """Query text should append order stages."""
     args = make_args(filter_tags=["work"], order_by_timestamp_desc=True, offset=5, max_results=10)
     argv = [
         "org",
@@ -133,12 +131,11 @@ def test_build_query_text_with_ordering_and_slice() -> None:
         "file.org",
     ]
 
-    query = _build_query_text(args, argv, include_ordering=True, include_slice=True)
+    query = _build_query_text(args, argv, include_ordering=True)
 
     assert query == (
-        '[ .[] | select(.tags[] matches "work") '
-        "| sort_by(.repeats + .deadline + .closed + .scheduled | max) ]"
-        "[$offset:($offset + $limit)]"
+        'select(.tags[] matches "work") '
+        "| sort_by(.repeats + .deadline + .closed + .scheduled | max)"
     )
 
 
@@ -147,12 +144,12 @@ def test_build_query_text_with_timestamp_asc_keeps_none_last() -> None:
     args = make_args(order_by_timestamp_asc=True)
     argv = ["org", "tasks", "list", "--order-by-timestamp-asc", "file.org"]
 
-    query = _build_query_text(args, argv, include_ordering=True, include_slice=False)
+    query = _build_query_text(args, argv, include_ordering=True)
 
     assert query == (
-        "[ .[] | sort_by(.repeats + .deadline + .closed + .scheduled | max) "
+        "sort_by(.repeats + .deadline + .closed + .scheduled | max) "
         "| reverse "
-        "| sort_by((.repeats + .deadline + .closed + .scheduled | max) != null) ]"
+        "| sort_by((.repeats + .deadline + .closed + .scheduled | max) != null)"
     )
 
 
@@ -161,9 +158,9 @@ def test_build_query_text_with_priority_ordering() -> None:
     args = make_args(order_by_priority=True)
     argv = ["org", "tasks", "list", "--order-by-priority", "file.org"]
 
-    query = _build_query_text(args, argv, include_ordering=True, include_slice=False)
+    query = _build_query_text(args, argv, include_ordering=True)
 
-    assert query == "[ .[] | sort_by(.priority) ]"
+    assert query == "sort_by(.priority)"
 
 
 def test_build_query_text_with_property_filter() -> None:
@@ -171,9 +168,9 @@ def test_build_query_text_with_property_filter() -> None:
     args = make_args(filter_properties=["priority=A"])
     argv = ["org", "stats", "all", "--filter-property", "priority=A", "file.org"]
 
-    query = _build_query_text(args, argv, include_ordering=False, include_slice=False)
+    query = _build_query_text(args, argv, include_ordering=False)
 
-    assert query == '[ .[] | select(.properties["priority"] == "A") ]'
+    assert query == 'select(.properties["priority"] == "A")'
 
 
 def test_build_query_text_with_builtin_with_tags_as_category_stage() -> None:
@@ -181,11 +178,9 @@ def test_build_query_text_with_builtin_with_tags_as_category_stage() -> None:
     args = make_args(with_tags_as_category=True)
     argv = ["org", "stats", "summary", "--with-tags-as-category", "file.org"]
 
-    query = _build_query_text(args, argv, include_ordering=False, include_slice=False)
+    query = _build_query_text(args, argv, include_ordering=False)
 
-    assert query == (
-        "[ .[] | (if .tags | length > 0 then .heading_category = .tags[0] else null); . ]"
-    )
+    assert query == "(if .tags | length > 0 then .heading_category = .tags[0] else null); ."
 
 
 def test_build_query_text_with_tags_as_category_default_appended_when_missing_in_argv() -> None:
@@ -193,11 +188,11 @@ def test_build_query_text_with_tags_as_category_default_appended_when_missing_in
     args = make_args(with_tags_as_category=True, filter_tags=["work"])
     argv = ["org", "stats", "summary", "--filter-tag", "work", "file.org"]
 
-    query = _build_query_text(args, argv, include_ordering=False, include_slice=False)
+    query = _build_query_text(args, argv, include_ordering=False)
 
     assert query == (
-        "[ .[] | (if .tags | length > 0 then .heading_category = .tags[0] else null); . "
-        '| select(.tags[] matches "work") ]'
+        "(if .tags | length > 0 then .heading_category = .tags[0] else null); . "
+        '| select(.tags[] matches "work")'
     )
 
 
@@ -206,12 +201,12 @@ def test_build_query_text_with_filter_completed() -> None:
     args = make_args(filter_completed=True)
     argv = ["org", "tasks", "list", "--filter-completed", "file.org"]
 
-    query = _build_query_text(args, argv, include_ordering=False, include_slice=False)
+    query = _build_query_text(args, argv, include_ordering=False)
 
     assert query == (
-        "[ .[] | select(if .repeats | length > 0 then .repeats | map(.is_completed) + "
+        "select(if .repeats | length > 0 then .repeats | map(.is_completed) + "
         "[.is_completed] | any else .is_completed) "
-        "| .repeats = [.repeats[] | select(.is_completed)]; .  ]"
+        "| .repeats = [.repeats[] | select(.is_completed)]; . "
     )
 
 
@@ -220,26 +215,26 @@ def test_build_query_text_with_filter_not_completed() -> None:
     args = make_args(filter_not_completed=True)
     argv = ["org", "tasks", "list", "--filter-not-completed", "file.org"]
 
-    query = _build_query_text(args, argv, include_ordering=False, include_slice=False)
+    query = _build_query_text(args, argv, include_ordering=False)
 
     assert query == (
-        "[ .[] | select(if .repeats | length > 0 then not(.repeats | map(.is_completed) + "
+        "select(if .repeats | length > 0 then not(.repeats | map(.is_completed) + "
         "[.is_completed] | any) else not(.is_completed)) "
-        "| .repeats = [.repeats[] | select(not(.is_completed))]; .  ]"
+        "| .repeats = [.repeats[] | select(not(.is_completed))]; . "
     )
 
 
-def test_build_query_logs_query_before_compile(caplog: pytest.LogCaptureFixture) -> None:
-    """build_query should log the query text before compilation."""
+def test_run_query_logs_query_before_compile(caplog: pytest.LogCaptureFixture) -> None:
+    """run_query should log the query text before compilation."""
     args = make_args(filter_tags=["simple"])
     argv = ["org", "stats", "all", "--filter-tag", "simple", "file.org"]
 
     org.logging.logger.setLevel(logging.INFO)
     org.logging.logger.propagate = True
     with caplog.at_level(logging.INFO, logger="org"):
-        _build_query(args, argv, include_ordering=False, include_slice=False)
+        _run_query(args, argv, include_ordering=False)
 
-    assert 'Query: [ .[] | select(.tags[] matches "simple") ]' in caplog.text
+    assert 'Query: select(.tags[] matches "simple")' in caplog.text
 
 
 def test_build_query_text_with_custom_filter_and_optional_arg() -> None:
@@ -266,13 +261,11 @@ def test_build_query_text_with_custom_filter_and_optional_arg() -> None:
         args,
         argv,
         include_ordering=False,
-        include_slice=False,
         config=config,
     )
 
     assert query == (
-        "[ .[] | let 3 as $arg in (select(.todo == $arg))"
-        " | let null as $arg in (select(.todo != null)) ]"
+        "let 3 as $arg in (select(.todo == $arg)) | let null as $arg in (select(.todo != null))"
     )
 
 
@@ -326,11 +319,10 @@ def test_build_query_text_custom_with_before_filters() -> None:
         args,
         argv,
         include_ordering=False,
-        include_slice=False,
         config=config,
     )
 
-    assert query.startswith('[ .[] | let "one" as $arg in (. + {"x": $arg}) | ')
+    assert query.startswith('let "one" as $arg in (. + {"x": $arg}) | ')
     assert "| let null as $arg in (select(.tag != null)) |" in query
 
 
@@ -353,13 +345,12 @@ def test_build_query_text_with_builtin_and_custom_with_stages_follow_argv_order(
         args,
         argv,
         include_ordering=False,
-        include_slice=False,
         config=config,
     )
 
     assert query == (
-        '[ .[] | let "one" as $arg in (. + {"x": $arg}) '
-        "| (if .tags | length > 0 then .heading_category = .tags[0] else null); . ]"
+        'let "one" as $arg in (. + {"x": $arg}) '
+        "| (if .tags | length > 0 then .heading_category = .tags[0] else null); ."
     )
 
 
@@ -374,11 +365,10 @@ def test_build_query_text_custom_ordering_for_stats() -> None:
         args,
         argv,
         include_ordering=False,
-        include_slice=False,
         config=config,
     )
 
-    assert query == "[ .[] | let null as $arg in (sort_by(.priority)) ]"
+    assert query == "let null as $arg in (sort_by(.priority))"
 
 
 def test_load_and_process_data_logs_query_context(caplog: pytest.LogCaptureFixture) -> None:
@@ -411,11 +401,11 @@ def test_build_query_text_preserves_mixed_ordering_cli_order() -> None:
         "file.org",
     ]
 
-    query = _build_query_text(args, argv, include_ordering=True, include_slice=False, config=config)
+    query = _build_query_text(args, argv, include_ordering=True, config=config)
 
     assert query == (
-        "[ .[] | let null as $arg in (sort_by(.priority))"
-        " | sort_by(.repeats + .deadline + .closed + .scheduled | max) ]"
+        "let null as $arg in (sort_by(.priority))"
+        " | sort_by(.repeats + .deadline + .closed + .scheduled | max)"
     )
 
 
@@ -425,7 +415,7 @@ def test_build_query_text_rejects_unknown_custom_switch() -> None:
     argv = ["org", "tasks", "list", "--filter-unknown", "file.org"]
 
     with pytest.raises(click.NoSuchOption, match="No such option"):
-        _build_query_text(args, argv, include_ordering=False, include_slice=False)
+        _build_query_text(args, argv, include_ordering=False)
 
 
 def test_build_query_text_custom_filter_requires_exactly_one_argument() -> None:
@@ -439,7 +429,7 @@ def test_build_query_text_custom_filter_requires_exactly_one_argument() -> None:
         typer.BadParameter,
         match="--filter-level-above requires exactly one argument",
     ):
-        _build_query_text(args, argv, include_ordering=False, include_slice=False, config=config)
+        _build_query_text(args, argv, include_ordering=False, config=config)
 
 
 def test_build_query_text_custom_order_by_requires_exactly_one_argument() -> None:
@@ -450,7 +440,7 @@ def test_build_query_text_custom_order_by_requires_exactly_one_argument() -> Non
     argv = ["org", "tasks", "list", "--order-by-weight"]
 
     with pytest.raises(typer.BadParameter, match="--order-by-weight requires exactly one argument"):
-        _build_query_text(args, argv, include_ordering=True, include_slice=False, config=config)
+        _build_query_text(args, argv, include_ordering=True, config=config)
 
 
 def test_build_query_text_custom_with_requires_exactly_one_argument() -> None:
@@ -461,7 +451,7 @@ def test_build_query_text_custom_with_requires_exactly_one_argument() -> None:
     argv = ["org", "tasks", "list", "--with-mark"]
 
     with pytest.raises(typer.BadParameter, match="--with-mark requires exactly one argument"):
-        _build_query_text(args, argv, include_ordering=False, include_slice=False, config=config)
+        _build_query_text(args, argv, include_ordering=False, config=config)
 
 
 def test_normalize_cli_files_consumes_required_custom_path_like_argument() -> None:
diff --git a/tests/commands/test_tasks_query_command.py b/tests/commands/test_tasks_query_command.py
index 6fe0af0..890a29b 100644
--- a/tests/commands/test_tasks_query_command.py
+++ b/tests/commands/test_tasks_query_command.py
@@ -60,7 +60,7 @@ def _make_args(files: list[str], query: str, **overrides: object) -> TasksQueryA
 def test_run_query_starts_from_root_nodes(capsys: pytest.CaptureFixture[str]) -> None:
     """Query should treat loaded root nodes as the initial stream."""
     fixture_path = os.path.join(FIXTURES_DIR, "multiple_tags.org")
-    args = _make_args([fixture_path], ".[] | .children | length")
+    args = _make_args([fixture_path], ".children | length")
 
     _run_tasks_query(args)
     captured = capsys.readouterr().out
@@ -73,7 +73,7 @@ def test_run_query_org_node_results_render_with_file_header(
 ) -> None:
     """Org node query results should render like detailed task output."""
     fixture_path = os.path.join(FIXTURES_DIR, "multiple_tags.org")
-    args = _make_args([fixture_path], ".[] | .children | .[0]")
+    args = _make_args([fixture_path], ".children | .[0]")
 
     _run_tasks_query(args)
     captured = capsys.readouterr().out
@@ -87,7 +87,7 @@ def test_run_query_org_root_results_render_with_file_header(
 ) -> None:
     """Org root query results should render as org syntax blocks."""
     fixture_path = os.path.join(FIXTURES_DIR, "multiple_tags.org")
-    args = _make_args([fixture_path], ".[]")
+    args = _make_args([fixture_path], ".")
 
     _run_tasks_query(args)
     captured = capsys.readouterr().out
@@ -102,14 +102,14 @@ def test_run_query_default_org_uses_plain_formatter_for_string_results(
 ) -> None:
     """Default org output should print plain lines for string-only results."""
     fixture_path = os.path.join(FIXTURES_DIR, "multiple_tags.org")
-    args = _make_args([fixture_path], ".[] | .children | .[] | .title_text")
+    args = _make_args([fixture_path], ".[] | .title_text")
 
-    def _fake_compiled_query(_stream: object, _context: object) -> list[object]:
+    def _fake_run_query(_inputs: object, _stages: object, _context: object) -> list[object]:
         return ["alpha", "beta"]
 
     monkeypatch.setattr(
-        "org.commands.tasks.query.compile_query_or_raise",
-        lambda _query, _error_builder: _fake_compiled_query,
+        "org.commands.tasks.query.run_query",
+        _fake_run_query,
     )
     monkeypatch.setattr(
         "org.commands.tasks.query.load_root_data",
@@ -130,12 +130,12 @@ def test_run_query_default_org_uses_json_formatter_for_mixed_results(
     fixture_path = os.path.join(FIXTURES_DIR, "multiple_tags.org")
     args = _make_args([fixture_path], ".[]")
 
-    def _fake_compiled_query(_stream: object, _context: object) -> list[object]:
+    def _fake_run_query(_inputs: object, _stages: object, _context: object) -> list[object]:
         return ["alpha", 1, None]
 
     monkeypatch.setattr(
-        "org.commands.tasks.query.compile_query_or_raise",
-        lambda _query, _error_builder: _fake_compiled_query,
+        "org.commands.tasks.query.run_query",
+        _fake_run_query,
     )
     monkeypatch.setattr(
         "org.commands.tasks.query.load_root_data",
@@ -154,14 +154,14 @@ def test_run_query_default_org_uses_json_formatter_for_none_result(
 ) -> None:
     """Default org output should print JSON null for non-org None results."""
     fixture_path = os.path.join(FIXTURES_DIR, "multiple_tags.org")
-    args = _make_args([fixture_path], ".[]")
+    args = _make_args([fixture_path], ".")
 
-    def _fake_compiled_query(_stream: object, _context: object) -> list[object]:
+    def _fake_run_query(_inputs: object, _stages: object, _context: object) -> list[object]:
         return [None]
 
     monkeypatch.setattr(
-        "org.commands.tasks.query.compile_query_or_raise",
-        lambda _query, _error_builder: _fake_compiled_query,
+        "org.commands.tasks.query.run_query",
+        _fake_run_query,
     )
     monkeypatch.setattr(
         "org.commands.tasks.query.load_root_data",
@@ -196,7 +196,7 @@ def test_query_empty_scheduled_timestamp_renders_json_null(
 ) -> None:
     """Querying an unset scheduled timestamp should render as JSON null."""
     fixture_path = os.path.join(FIXTURES_DIR, "simple.org")
-    args = _make_args([fixture_path], ".[][].scheduled")
+    args = _make_args([fixture_path], ".[] | .scheduled")
 
     _run_tasks_query(args)
     captured = capsys.readouterr().out
@@ -246,7 +246,7 @@ def _fake_run_tasks_query(
 
     monkeypatch.setattr("org.commands.tasks.query.run_tasks_query", _fake_run_tasks_query)
 
-    result = runner.invoke(configured_app, ["tasks", "query", ".[]", fixture_path])
+    result = runner.invoke(configured_app, ["tasks", "query", ".", fixture_path])
 
     assert result.exit_code == 0
     assert captured == {"offset": 2, "width": 65}
@@ -269,7 +269,7 @@ def test_run_query_markdown_converts_org_results(
 ) -> None:
     """Markdown query formatter should invoke pandoc with markdown output."""
     fixture_path = os.path.join(FIXTURES_DIR, "multiple_tags.org")
-    args = _make_args([fixture_path], ".[] | .children | .[]", out="markdown")
+    args = _make_args([fixture_path], ".children | .[]", out="markdown")
     seen: dict[str, object] = {}
 
     def _fake_pandoc(org_text: str, output_format: str, pandoc_args: list[str]) -> str:
@@ -294,7 +294,7 @@ def test_run_query_markdown_converts_scalar_results(
 ) -> None:
     """Markdown query formatter should pass scalar outputs to pandoc."""
     fixture_path = os.path.join(FIXTURES_DIR, "multiple_tags.org")
-    args = _make_args([fixture_path], ".[] | .children | length", out="markdown")
+    args = _make_args([fixture_path], ".children | length", out="markdown")
     seen: dict[str, object] = {}
 
     def _fake_pandoc(org_text: str, output_format: str, pandoc_args: list[str]) -> str:
@@ -320,7 +320,7 @@ def test_run_query_accepts_arbitrary_pandoc_output_format(
     fixture_path = os.path.join(FIXTURES_DIR, "multiple_tags.org")
     args = _make_args(
         [fixture_path],
-        ".[] | .children | .[]",
+        ".children | .[]",
         out="gfm",
         pandoc_args="--wrap=none",
     )
@@ -350,7 +350,7 @@ def prepare(
             raise OutputFormatError("pandoc missing")
 
     fixture_path = os.path.join(FIXTURES_DIR, "multiple_tags.org")
-    args = _make_args([fixture_path], ".[]", out="markdown")
+    args = _make_args([fixture_path], ".", out="markdown")
     monkeypatch.setattr(
         "org.commands.tasks.query.get_query_formatter",
         lambda _out, _pandoc_args: _FailingFormatter(),
@@ -372,7 +372,7 @@ def _should_not_call(_org_text: str, _output_format: str, _pandoc_args: list[str
     fixture_path = os.path.join(FIXTURES_DIR, "multiple_tags.org")
     args = _make_args(
         [fixture_path],
-        ".[] | .children | .[] | select(false)",
+        ".children | .[] | select(false)",
         out="gfm",
     )
     monkeypatch.setattr("org.commands.tasks.query._org_to_pandoc_format", _should_not_call)
@@ -386,7 +386,7 @@ def _should_not_call(_org_text: str, _output_format: str, _pandoc_args: list[str
 def test_run_query_json_root_result_is_single_object(capsys: pytest.CaptureFixture[str]) -> None:
     """JSON query output should return one object for a single root result."""
     fixture_path = os.path.join(FIXTURES_DIR, "multiple_tags.org")
-    args = _make_args([fixture_path], ".[]", out=OutputFormat.JSON)
+    args = _make_args([fixture_path], ".", out=OutputFormat.JSON)
 
     _run_tasks_query(args)
     captured = capsys.readouterr().out
@@ -401,7 +401,7 @@ def test_run_query_json_root_result_is_single_object(capsys: pytest.CaptureFixtu
 def test_run_query_json_node_result_excludes_env(capsys: pytest.CaptureFixture[str]) -> None:
     """JSON query output should exclude env from non-root org nodes."""
     fixture_path = os.path.join(FIXTURES_DIR, "multiple_tags.org")
-    args = _make_args([fixture_path], ".[] | .children | .[0]", out=OutputFormat.JSON)
+    args = _make_args([fixture_path], ".children | .[0]", out=OutputFormat.JSON)
 
     _run_tasks_query(args)
     captured = capsys.readouterr().out
@@ -415,7 +415,7 @@ def test_run_query_json_node_result_excludes_env(capsys: pytest.CaptureFixture[s
 def test_run_query_json_scalars_emit_single_json_value(capsys: pytest.CaptureFixture[str]) -> None:
     """JSON query output should emit scalar JSON when one item remains."""
     fixture_path = os.path.join(FIXTURES_DIR, "multiple_tags.org")
-    args = _make_args([fixture_path], ".[] | .children | length", out=OutputFormat.JSON)
+    args = _make_args([fixture_path], ".children | length", out=OutputFormat.JSON)
 
     _run_tasks_query(args)
     captured = capsys.readouterr().out
@@ -432,7 +432,7 @@ def test_run_query_json_preserves_multiple_collection_results(
         os.path.join(FIXTURES_DIR, "multiple_tags.org"),
         os.path.join(FIXTURES_DIR, "simple.org"),
     ]
-    args = _make_args(fixture_paths, ".[] | .children", out=OutputFormat.JSON)
+    args = _make_args(fixture_paths, ".children", out=OutputFormat.JSON)
 
     _run_tasks_query(args)
     captured = capsys.readouterr().out
@@ -463,7 +463,7 @@ def test_is_org_object_supports_org_parser_text_and_element_types() -> None:
 def test_run_query_invalid_pandoc_args_is_usage_error() -> None:
     """Malformed pandoc args should be surfaced as a CLI usage error."""
     fixture_path = os.path.join(FIXTURES_DIR, "multiple_tags.org")
-    args = _make_args([fixture_path], ".[]", out="gfm", pandoc_args='"')
+    args = _make_args([fixture_path], ".", out="gfm", pandoc_args='"')
 
     with pytest.raises(click.UsageError, match="No closing quotation"):
         _run_tasks_query(args)
diff --git a/tests/test_query_language_compiler.py b/tests/test_query_language_compiler.py
index 9b7263c..2513098 100644
--- a/tests/test_query_language_compiler.py
+++ b/tests/test_query_language_compiler.py
@@ -2,11 +2,18 @@
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
+import org.query.engine.parser as parser_module
 from org.query.engine.ast import NumberLiteral
 from org.query.engine.compiler import compile_expr, compile_query_text
 from org.query.engine.interpreter import EvalContext, Stream
 
 
+if TYPE_CHECKING:
+    import pytest
+
+
 def test_compile_expr_returns_executable_callable() -> None:
     """compile_expr should execute the provided AST expression."""
     compiled = compile_expr(NumberLiteral(42))
@@ -19,3 +26,22 @@ def test_compile_query_text_parses_and_executes_expression() -> None:
     compiled = compile_query_text(".[] | .title_text")
     result = compiled(Stream([[{"title_text": "Task"}]]), EvalContext({}))
     assert result == ["Task"]
+
+
+def test_parse_query_memoizes_repeated_queries(monkeypatch: pytest.MonkeyPatch) -> None:
+    """parse_query should only parse identical query text once per process."""
+    parser_module.parse_query.cache_clear()
+    calls = 0
+    original_parse = parser_module.QUERY_PARSER.parse
+
+    def counted_parse(query: str) -> object:
+        nonlocal calls
+        calls += 1
+        return original_parse(query)
+
+    monkeypatch.setattr(parser_module.QUERY_PARSER, "parse", counted_parse)
+
+    compile_query_text(".[] | .title_text")
+    compile_query_text(".[] | .title_text")
+
+    assert calls == 1
diff --git a/tests/test_query_language_parser.py b/tests/test_query_language_parser.py
index c37e4ad..ce8f0bb 100644
--- a/tests/test_query_language_parser.py
+++ b/tests/test_query_language_parser.py
@@ -69,6 +69,7 @@
         "str(.title_text)",
         'int("42")',
         'float("3.14")',
+        "fold",
         'bool("true")',
         'ts("<2026-03-01 Sun 10:00-12:00>")',
         "sha256",
@@ -164,6 +165,14 @@ def test_parse_fold_shape() -> None:
     assert isinstance(expr, Fold)
 
 
+def test_parse_fold_function_shape() -> None:
+    """Parser should parse fold built-in as a no-arg function call."""
+    expr = parse_query("fold")
+    assert isinstance(expr, FunctionCall)
+    assert expr.name == "fold"
+    assert expr.argument is None
+
+
 def test_parse_dict_assignment_shape() -> None:
     """Parser should parse dictionary assignment expressions."""
     expr = parse_query('.properties["done"] = true')
diff --git a/tests/test_query_language_runtime.py b/tests/test_query_language_runtime.py
index 3f2bf4f..385f6e9 100644
--- a/tests/test_query_language_runtime.py
+++ b/tests/test_query_language_runtime.py
@@ -669,6 +669,19 @@ def test_runtime_fold_operator() -> None:
     assert empty_fold == [[]]
 
 
+def test_runtime_fold_function_collects_current_stream() -> None:
+    """fold should collect the current stream into one list value."""
+    result = _execute("[1, 2, 3][] | fold", [None], None)
+    assert result == [[1, 2, 3]]
+
+
+def test_runtime_fold_function_collects_substream_before_later_stages() -> None:
+    """fold should collect the current intermediate stream before later stages consume it."""
+    nodes = _sample_nodes()
+    result = _execute(".[] | (.children[] | .title_text | fold) | length", nodes, None)
+    assert result == [2]
+
+
 def test_runtime_returns_stream_type() -> None:
     """Compiled queries should return Stream instances."""
     compiled = compile_query_text(".[] | .title_text")