@@ -44,7 +44,7 @@
 from comfy_cli._safe_exec import resolve_required_binary
 from comfy_cli.command.pack_scan import iter_pack_dirs as _iter_pack_dirs
 from comfy_cli.command.pack_scan import read_pyproject as _read_pyproject
-from comfy_cli.file_utils import atomic_write_text
+from comfy_cli.file_utils import atomic_write_text, cache_dir
 from comfy_cli.registry import RegistryAPI
 
 CACHE_TTL_SECONDS = 3600  # 1 hour
@@ -67,9 +67,8 @@
 
 
 def _cache_path() -> Path:
-    """Where the latest-version cache lives. XDG-respecting."""
-    base = os.environ.get("XDG_CACHE_HOME") or os.path.expanduser("~/.cache")
-    return Path(base) / "comfy-cli" / "outdated.json"
+    """Where the latest-version cache lives."""
+    return cache_dir() / "outdated.json"
 
 
 def _load_cache() -> dict[str, Any]:
 
@@ -31,7 +31,7 @@
 import typer
 
 from comfy_cli import knowledge, tracking, workflow_ops
-from comfy_cli.file_utils import atomic_write_bytes
+from comfy_cli.file_utils import atomic_write_bytes, cache_dir
 from comfy_cli.http import ResponseTooLarge, plain_urlopen, read_capped
 from comfy_cli.output import get_renderer, rprint
 
@@ -80,9 +80,8 @@
 
 
 def _cache_path() -> Path:
-    """Where the gallery index lives on disk. XDG-respecting."""
-    base = os.environ.get("XDG_CACHE_HOME") or os.path.expanduser("~/.cache")
-    return Path(base) / "comfy-cli" / "gallery" / "index.json"
+    """Where the gallery index lives on disk."""
+    return cache_dir() / "gallery" / "index.json"
 
 
 def _fetch_gallery(url: str = GALLERY_URL, timeout: float = 15.0) -> bytes:
@@ -358,7 +357,17 @@ def _spawn_background_refresh() -> bool:
     # *non-atomic* config.ini rewrite — racing the foreground process and risking
     # a corrupt config. `_refresh-cache` is contractually 'no telemetry,
     # best-effort', so opt the child out of consent entirely.
-    child_env = {**os.environ, "COMFY_NO_TELEMETRY": "1", "DO_NOT_TRACK": "1"}
+    # Pin the child to the cache dir *this* process resolved, not a raw
+    # `COMFY_CACHE_DIR` value: the child's cwd is `_refresh_cwd()`, already
+    # inside that dir, so re-resolving a relative override there would nest a
+    # second cache tree under the first and the foreground cache would never
+    # see the refresh.
+    child_env = {
+        **os.environ,
+        "COMFY_CACHE_DIR": str(cache_dir()),
+        "COMFY_NO_TELEMETRY": "1",
+        "DO_NOT_TRACK": "1",
+    }
 
     kwargs: dict[str, Any] = dict(
         stdout=subprocess.DEVNULL,
@@ -1214,9 +1223,8 @@ def _template_workflow_cache_path(name: str) -> Path:
     ``quote(..., safe="")`` turns any ``/`` or ``..`` segment into ``%2F``/``..``
     with no path separators, keeping the file strictly inside ``gallery/templates``.
     """
-    base = os.environ.get("XDG_CACHE_HOME") or os.path.expanduser("~/.cache")
     safe_name = urllib.parse.quote(name, safe="")
-    return Path(base) / "comfy-cli" / "gallery" / "templates" / f"{safe_name}.json"
+    return cache_dir() / "gallery" / "templates" / f"{safe_name}.json"
 
 
 def _iter_workflow_nodes(wf: dict[str, Any]):
 
@@ -61,6 +61,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import json
 import os
 import tempfile
@@ -71,6 +72,7 @@
 from pathlib import Path
 from typing import Any
 
+from comfy_cli.file_utils import atomic_write_text, cache_dir
 from comfy_cli.http import plain_urlopen, read_capped
 
 # Public source of truth. Both files are at the repo root on ``main``.
@@ -101,6 +103,9 @@
 # miss, so an old-format file is re-fetched rather than misread.
 _CACHE_SCHEMA = 1
 
+# Bump when parse_supported_nodes / parse_disable_config change their output shape.
+_PARSED_SCHEMA = 1
+
 # These are two small YAML documents (~32 KB bundled). The shared 64 MiB default
 # is a ceiling for ``/object_info``-sized payloads; a much tighter cap here means
 # a misbehaving or hostile upstream can't stream unbounded data into the memory
@@ -119,9 +124,17 @@ def network_disabled() -> bool:
     return os.environ.get("COMFY_CLI_NO_REMOTE_REFRESH", "").strip().lower() not in _FALSEY
 
 
+def index_cache_disabled() -> bool:
+    """True when ``COMFY_NO_CACHE`` opts out of the parsed-annotation cache.
+
+    Scope is :func:`parsed_annotations` only: the raw YAML pair cache and the
+    object_info cache are unaffected.
+    """
+    return os.environ.get("COMFY_NO_CACHE", "").strip().lower() not in _FALSEY
+
+
 def _cache_dir() -> Path:
-    base = os.environ.get("XDG_CACHE_HOME") or os.path.expanduser("~/.cache")
-    return Path(base) / "comfy-cli" / "comfy-complete"
+    return cache_dir() / "comfy-complete"
 
 
 def _cache_file() -> Path:
@@ -328,10 +341,10 @@ def _drop_legacy_cache() -> None:
     Harmless if left behind — nothing reads it any more — but it's dead bytes in
     the user's cache dir, so clean it up the first time we write the new one.
     """
-    cache_dir = _cache_dir()
+    directory = _cache_dir()
     for filename in _FILES:
         try:
-            (cache_dir / filename).unlink()
+            (directory / filename).unlink()
         except OSError:
             pass
 
@@ -351,18 +364,33 @@ def _read_cached_pair(*, require_fresh: bool) -> dict[str, bytes] | None:
     """The cached pair, if it is complete, valid, and (optionally) fresh.
 
     Returns ``None`` on anything short of that — missing file, unreadable,
-    non-JSON, wrong schema, an absent entry, or a body that fails its validator.
-    A cache written by an older comfy-cli (per-file, pre-validation) lands in
-    that bucket too, so it is simply re-fetched rather than trusted. Every
-    ``None`` hands the decision back to the caller's next fallback, which
-    eventually reaches the bundled snapshot — never a silent blank annotation.
+    oversized, non-JSON, wrong schema, an absent entry, or (for a pair not
+    already vouched for below) a body that fails its validator. A cache written by an
+    older comfy-cli (per-file, pre-validation) lands in that bucket too, so it
+    is simply re-fetched rather than trusted. Every ``None`` hands the
+    decision back to the caller's next fallback, which eventually reaches the
+    bundled snapshot — never a silent blank annotation.
+
+    ``_persist_pair`` only ever writes a pair that passed ``_VALIDATORS`` in
+    ``_fetch_one``, so a *freshly written* file is valid by construction. But
+    ``COMFY_CACHE_DIR`` makes this file easy to point at or hand-edit, so that
+    guarantee doesn't hold forever. The full YAML validation this used to run
+    on every read is skipped only once ``parsed_annotations`` has already
+    parsed this exact pair successfully — its cache file existing on disk is
+    proof the bytes are good — so the hot path stays cheap while a poisoned
+    cache is still caught the first time it's seen.
     """
     cache = _cache_file()
     if require_fresh and not _is_fresh(cache):
         return None
     try:
+        # Same reasoning as ``_read_parsed``'s size cap: this path isn't even
+        # digest-derived, so anything able to write the cache dir can plant an
+        # arbitrarily large or deeply-nested file here.
+        if cache.stat().st_size > _MAX_ANNOTATION_BYTES:
+            return None
         payload = json.loads(cache.read_bytes())
-    except (OSError, ValueError):
+    except (OSError, ValueError, RecursionError):
         return None
     if not isinstance(payload, dict) or payload.get("schema") != _CACHE_SCHEMA:
         return None
@@ -375,13 +403,125 @@ def _read_cached_pair(*, require_fresh: bool) -> dict[str, bytes] | None:
         text = files.get(filename)
         if not isinstance(text, str):
             return None
-        data = text.encode("utf-8")
-        if not _VALIDATORS[filename](data):
-            return None
-        out[filename] = data
+        out[filename] = text.encode("utf-8")
+
+    if not _parsed_cache_path(out[_SUPPORTED_NODES], out[_CLOUD_DISABLE]).exists():
+        for filename in _FILES:
+            if not _VALIDATORS[filename](out[filename]):
+                return None
     return out
 
 
+# ---------------------------------------------------------------------------
+# Parsed-annotation cache
+# ---------------------------------------------------------------------------
+
+_PARSED_PREFIX = "annotations-parsed-"
+_PARSED_GLOB = f"{_PARSED_PREFIX}*.json"
+
+ParsedAnnotations = tuple[dict[str, str], dict[str, list[str]], set[str]]
+
+
+def _parsed_cache_path(sup: bytes, dis: bytes) -> Path:
+    key = hashlib.sha256(sup + b"\0" + dis).hexdigest()[:16]
+    return _cache_dir() / f"{_PARSED_PREFIX}v{_PARSED_SCHEMA}-{key}.json"
+
+
+def _read_parsed(path: Path) -> ParsedAnnotations | None:
+    try:
+        # The filename is a predictable digest over public upstream YAML, so
+        # anything able to write the cache dir can plant a huge file at the
+        # exact path this will read; cap it the same way network reads are
+        # capped rather than handing an unbounded read to ``json.loads``.
+        if path.stat().st_size > _MAX_ANNOTATION_BYTES:
+            return None
+        payload = json.loads(path.read_bytes())
+        node_pack = payload["node_pack"]
+        node_labels = payload["node_labels"]
+        disable_labels = payload["disable_labels"]
+        if not (
+            isinstance(node_pack, dict)
+            and isinstance(node_labels, dict)
+            and isinstance(disable_labels, list)
+            and all(isinstance(pack, str) for pack in node_pack.values())
+            and all(
+                isinstance(labels, list) and all(isinstance(label, str) for label in labels)
+                for labels in node_labels.values()
+            )
+            and all(isinstance(label, str) for label in disable_labels)
+        ):
+            return None
+        return node_pack, node_labels, set(disable_labels)
+    except (OSError, ValueError, KeyError, TypeError, RecursionError):
+        return None
+
+
+def _write_parsed(path: Path, parsed: ParsedAnnotations) -> None:
+    """Persist ``parsed`` at ``path``, evicting every other parsed file after.
+
+    Newest-only keeps the cache dir from accumulating one file per annotation
+    generation. Writing first and evicting siblings afterward means a failed
+    write leaves the previous good entry in place instead of an empty dir.
+    Best-effort throughout.
+    """
+    node_pack, node_labels, disable_labels = parsed
+    # ``json.dumps`` coerces a non-str dict key (int, float, bool, None) to a
+    # string instead of raising, which would silently change what a later
+    # cache hit returns compared to this cache miss. Treat it like any other
+    # unserialisable result: skip the write, keep re-parsing.
+    if not (all(isinstance(k, str) for k in node_pack) and all(isinstance(k, str) for k in node_labels)):
+        return
+    try:
+        blob = json.dumps(
+            {"node_pack": node_pack, "node_labels": node_labels, "disable_labels": sorted(disable_labels)}
+        )
+        atomic_write_text(path, blob)
+    except (OSError, TypeError, ValueError):
+        return
+    for sibling in path.parent.glob(_PARSED_GLOB):
+        if sibling != path:
+            try:
+                sibling.unlink()
+            except OSError:
+                pass
+
+
+def parsed_annotations(
+    supported_nodes_yaml: bytes | None,
+    cloud_disable_yaml: bytes | None,
+) -> ParsedAnnotations:
+    """``(node_pack, node_labels, disable_labels)`` for the given annotation bytes.
+
+    The parse result is stored on disk as JSON keyed by a digest of the input
+    bytes, so the YAML is parsed once per upstream generation rather than on
+    every call. A new pair gets a new key and the previous file is evicted on
+    write. ``COMFY_NO_CACHE`` skips the disk entirely. Never raises: a missing,
+    unreadable or malformed cache file is a miss and the documents are parsed
+    again.
+    """
+    sup = supported_nodes_yaml or b""
+    dis = cloud_disable_yaml or b""
+    if not sup and not dis:
+        return {}, {}, set()
+    use_cache = not index_cache_disabled()
+    path = _parsed_cache_path(sup, dis) if use_cache else None
+
+    if path is not None:
+        cached = _read_parsed(path)
+        if cached is not None:
+            return cached
+
+    from comfy_cli.cql.engine import parse_disable_config, parse_supported_nodes
+
+    node_pack, node_labels = parse_supported_nodes(sup) if sup else ({}, {})
+    disable_labels = parse_disable_config(dis) if dis else set()
+    parsed: ParsedAnnotations = (node_pack, node_labels, disable_labels)
+
+    if path is not None:
+        _write_parsed(path, parsed)
+    return parsed
+
+
 def _failure_stamp() -> Path:
     return _cache_dir() / _FAILURE_STAMP
 
 
@@ -866,14 +866,11 @@ def annotate(
         supported_nodes_yaml: bytes | None = None,
         cloud_disable_yaml: bytes | None = None,
     ) -> None:
-        node_pack: dict[str, str] = {}
-        node_labels: dict[str, list[str]] = {}
-        disable_labels: set[str] = set()
-
-        if supported_nodes_yaml:
-            node_pack, node_labels = parse_supported_nodes(supported_nodes_yaml)
-        if cloud_disable_yaml:
-            disable_labels = parse_disable_config(cloud_disable_yaml)
+        from comfy_cli.cql import annotations_source
+
+        node_pack, node_labels, disable_labels = annotations_source.parsed_annotations(
+            supported_nodes_yaml, cloud_disable_yaml
+        )
 
         for nid, m in self._nodes.items():
             if nid in node_pack:
 
@@ -29,7 +29,7 @@
 from typing import Any
 
 from comfy_cli.cql.errors import CQLRuntimeError
-from comfy_cli.file_utils import atomic_write_text
+from comfy_cli.file_utils import atomic_write_text, cache_dir
 
 # ---- normalization --------------------------------------------------------
 
@@ -196,19 +196,6 @@ def _from_api_workflow(data: dict[str, Any]) -> dict[str, Any]:
 # An explicit ``--input <object_info.json>`` always wins and is never cached.
 
 
-def _cache_dir() -> Path:
-    """Return the per-user cache directory for comfy-cli object_info dumps.
-
-    Honors ``XDG_CACHE_HOME`` (Linux/freedesktop convention) and falls back to
-    ``~/.cache/comfy-cli`` everywhere else. We deliberately use a plain cache
-    dir rather than the config dir: this data is reconstructible and safe to
-    delete at any time.
-    """
-    xdg = os.environ.get("XDG_CACHE_HOME")
-    base = Path(xdg) if xdg else Path.home() / ".cache"
-    return base / "comfy-cli"
-
-
 def _host_key_digest(host_key: str) -> str:
     """Short, filesystem-safe hash of the target identity.
 
@@ -221,7 +208,7 @@ def _host_key_digest(host_key: str) -> str:
 
 def object_info_cache_path(host_key: str) -> Path:
     """Cache-file path for a given target identity."""
-    return _cache_dir() / f"object_info-{_host_key_digest(host_key)}.json"
+    return cache_dir() / f"object_info-{_host_key_digest(host_key)}.json"
 
 
 def write_object_info_cache(host_key: str, data: dict[str, Any]) -> None:
 
@@ -20,6 +20,27 @@
 
 logger = logging.getLogger(__name__)
 
+
+def cache_dir() -> pathlib.Path:
+    """comfy-cli's per-user cache root.
+
+    ``COMFY_CACHE_DIR`` wins; else ``$XDG_CACHE_HOME/comfy-cli``; else ``~/.cache/comfy-cli``.
+
+    Both env vars are resolved with ``os.path.expanduser`` (never raises, unlike
+    ``pathlib.Path.expanduser``, when the home directory can't be determined —
+    e.g. no ``HOME`` and no passwd entry, routine in a container) and made
+    absolute, so the result doesn't depend on the calling process's cwd. That
+    matters because a relative override would otherwise resolve differently in
+    every process — including a background refresher that launches its child
+    from inside this very directory.
+    """
+    explicit = os.environ.get("COMFY_CACHE_DIR", "").strip()
+    if explicit:
+        return pathlib.Path(os.path.abspath(os.path.expanduser(explicit)))
+    base = os.environ.get("XDG_CACHE_HOME", "").strip() or os.path.expanduser("~/.cache")
+    return pathlib.Path(os.path.abspath(os.path.expanduser(base))) / "comfy-cli"
+
+
 # ---------------------------------------------------------------------------
 # Atomic writes — the write policy
 # ---------------------------------------------------------------------------