Fix stale AST cache crash in fragile-path lint (#120)

* Refresh README with clearer positioning

* Fix stale AST cache crash in fragile-path lint

Author: lbliii

src/kida/bytecode_cache.py

@@ -37,9 +37,11 @@
 import logging
 import marshal
 import pickle
+import re
 import struct
 import sys
 import time
+from importlib.metadata import PackageNotFoundError, version
 from pathlib import Path
 from tempfile import NamedTemporaryFile
 from typing import TYPE_CHECKING, cast
@@ -49,6 +51,7 @@
 logger = logging.getLogger("kida.bytecode_cache")
 
 if TYPE_CHECKING:
+    from collections.abc import Iterable
     from types import CodeType
 
     from kida.nodes.base import Node
@@ -66,6 +69,15 @@
 # Python version tag for cache invalidation across Python upgrades
 _PY_VERSION_TAG = f"py{sys.version_info.major}{sys.version_info.minor}"
 
+# Cache ABI tag for Kida compiler/AST changes. Bytecode cache entries contain
+# compiled Python code and, optionally, pickled Kida AST nodes; both are tied to
+# the engine version and to AST dataclass shape.
+try:
+    _KIDA_VERSION_TAG = re.sub(r"[^A-Za-z0-9]+", "_", version("kida-templates")).strip("_")
+except PackageNotFoundError:  # pragma: no cover - package metadata may be absent in ad-hoc embeds
+    _KIDA_VERSION_TAG = "unknown"
+_CACHE_ABI_TAG = f"{_PY_VERSION_TAG}_kida{_KIDA_VERSION_TAG}_ast2"
+
 
 class BytecodeCache:
     """Persist compiled template bytecode to disk.
@@ -122,7 +134,7 @@ def _make_path(self, name: str, source_hash: str, context_hash: str | None = Non
         if context_hash:
             hash_key = f"{hash_key}_{context_hash[:16]}"
         filename = self._pattern.format(
-            version=_PY_VERSION_TAG,
+            version=_CACHE_ABI_TAG,
             name=safe_name,
             hash=hash_key,
         )
@@ -221,6 +233,11 @@ def get(
                             "Bytecode cache: failed to unpickle AST for '%s': %s", name, exc
                         )
                         ast = None
+                    if ast is not None and not _cached_ast_is_compatible(ast):
+                        logger.debug("Bytecode cache: incompatible cached AST for '%s'", name)
+                        with contextlib.suppress(OSError):
+                            path.unlink(missing_ok=True)
+                        return None, None, None
 
                 return code, ast, precomputed
 
@@ -252,6 +269,11 @@ def get(
                             "Bytecode cache: failed to unpickle AST (v2) for '%s': %s", name, exc
                         )
                         ast = None
+                    if ast is not None and not _cached_ast_is_compatible(ast):
+                        logger.debug("Bytecode cache: incompatible cached AST (v2) for '%s'", name)
+                        with contextlib.suppress(OSError):
+                            path.unlink(missing_ok=True)
+                        return None, None, None
 
                 return code, ast, None
             else:
@@ -408,3 +430,39 @@ def stats(self) -> dict[str, int]:
 def hash_source(source: str) -> str:
     """Generate hash of template source for cache key."""
     return hashlib.sha256(source.encode()).hexdigest()
+
+
+def _cached_ast_is_compatible(ast: object) -> bool:
+    """Return True when an unpickled AST matches the current dataclass shape."""
+    from kida.nodes.base import Node
+
+    if not isinstance(ast, Node):
+        return False
+
+    stack: list[Node] = [ast]
+    while stack:
+        node = stack.pop()
+        for field_name in node.__dataclass_fields__:
+            try:
+                value = getattr(node, field_name)
+            except AttributeError:
+                return False
+            if isinstance(value, Node):
+                stack.append(value)
+            elif isinstance(value, (list, tuple)):
+                _extend_node_sequence(stack, value)
+            elif isinstance(value, dict):
+                _extend_node_sequence(stack, value.values())
+    return True
+
+
+def _extend_node_sequence(stack: list[Node], seq: Iterable[object]) -> None:
+    from kida.nodes.base import Node
+
+    for item in seq:
+        if isinstance(item, Node):
+            stack.append(item)
+        elif isinstance(item, (list, tuple)):
+            _extend_node_sequence(stack, item)
+        elif isinstance(item, dict):
+            _extend_node_sequence(stack, item.values())

src/kida/compiler/partial_eval.py

@@ -26,7 +26,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable, Mapping, Sequence
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from typing import Any, final
 
 from kida.nodes import (
@@ -1904,14 +1904,7 @@ def _transform_expr(self, expr: Expr) -> Expr:
                 new_value = self._transform_expr(expr.value)
                 if new_value is expr.value:
                     return expr
-                return Filter(
-                    lineno=expr.lineno,
-                    col_offset=expr.col_offset,
-                    name=expr.name,
-                    value=new_value,
-                    args=expr.args,
-                    kwargs=expr.kwargs,
-                )
+                return replace(expr, value=new_value)
 
             case Pipeline():
                 val = self._try_eval(expr)

tests/test_fragile_paths_lint.py

@@ -2,11 +2,15 @@
 
 from __future__ import annotations
 
+import pickle
+import struct
 from typing import TYPE_CHECKING
 
 from kida import DictLoader, Environment
 from kida.analysis.fragile_paths import check_fragile_paths
+from kida.bytecode_cache import BytecodeCache, hash_source
 from kida.cli import main
+from kida.nodes import Filter, Name
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -20,6 +24,24 @@ def _ast(source: str, caller: str) -> object:
     return tpl._optimized_ast
 
 
+def _stale_filter_pickle() -> bytes:
+    current = bytearray(
+        pickle.dumps(
+            Filter(
+                lineno=1,
+                col_offset=35,
+                value=Name(lineno=1, col_offset=35, name="route_link_attrs"),
+                name="html_attrs",
+            ),
+            protocol=5,
+        )
+    )
+    del current[-4]  # Drop NEWFALSE for Filter.parenthesized from the state list.
+    frame_len = struct.unpack("<Q", current[3:11])[0]
+    current[3:11] = struct.pack("<Q", frame_len - 1)
+    return bytes(current)
+
+
 def test_same_folder_include_is_flagged() -> None:
     ast = _ast('{% include "pages/card.html" %}', "pages/about.html")
     issues = check_fragile_paths(ast, "pages/about.html")
@@ -126,6 +148,29 @@ def test_cli_lint_fragile_paths_exit_nonzero(
     assert "./card.html" in err
 
 
+def test_cli_lint_fragile_paths_ignores_stale_cached_ast(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    """A stale AST cache entry should recompile, not crash the lint visitor."""
+    (tmp_path / "pages").mkdir()
+    source = '{% include "pages/card.html" %}{{ href | upper }}'
+    (tmp_path / "pages" / "about.html").write_text(source)
+    (tmp_path / "pages" / "card.html").write_text("CARD")
+
+    cache = BytecodeCache(tmp_path / "__pycache__" / "kida")
+    source_hash = hash_source(source)
+    cache.set("pages/about.html", source_hash, compile("x = 1", "<stale>", "exec"))
+    path = cache._make_path("pages/about.html", source_hash)
+    path.write_bytes(path.read_bytes() + _stale_filter_pickle())
+
+    exit_code = main(["check", str(tmp_path), "--lint-fragile-paths"])
+    assert exit_code == 1
+
+    err = capsys.readouterr().err
+    assert "lint/fragile-path" in err
+    assert "Traceback" not in err
+
+
 def test_cli_lint_flag_default_off(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
     """Without the flag the lint is silent — default `kida check` is unchanged."""
     (tmp_path / "pages").mkdir()

tests/test_kida_bytecode_cache.py

@@ -3,6 +3,8 @@
 Tests persistent template caching for fast cold-start.
 """
 
+import pickle
+import struct
 import time
 from concurrent.futures import ThreadPoolExecutor
 
@@ -12,6 +14,27 @@
 from kida.exceptions import TemplateNotFoundError
 
 
+def _stale_filter_pickle() -> bytes:
+    """Return a Filter pickle from before the parenthesized slot existed."""
+    from kida.nodes import Filter, Name
+
+    current = bytearray(
+        pickle.dumps(
+            Filter(
+                lineno=1,
+                col_offset=3,
+                value=Name(lineno=1, col_offset=3, name="route_link_attrs"),
+                name="html_attrs",
+            ),
+            protocol=5,
+        )
+    )
+    del current[-4]  # Drop NEWFALSE for Filter.parenthesized from the state list.
+    frame_len = struct.unpack("<Q", current[3:11])[0]
+    current[3:11] = struct.pack("<Q", frame_len - 1)
+    return bytes(current)
+
+
 class TestBytecodeCache:
     """Tests for bytecode cache."""
 
@@ -255,6 +278,18 @@ def test_corrupted_cache_file_handled(self, cache, cache_dir):
         # Corrupted file should be removed
         assert not corrupted_path.exists()
 
+    def test_incompatible_cached_ast_is_discarded(self, cache):
+        """Stale pickled AST nodes from an older Kida schema do not crash."""
+        source_hash = hash_source("{{ route_link_attrs(href) | html_attrs }}")
+        code = compile("x = 1", "<test>", "exec")
+        cache.set("stale.html", source_hash, code)
+        path = cache._make_path("stale.html", source_hash)
+        path.write_bytes(path.read_bytes() + _stale_filter_pickle())
+
+        assert cache.stats()["file_count"] == 1
+        assert cache.get("stale.html", source_hash) == (None, None, None)
+        assert cache.stats()["file_count"] == 0
+
 
 class TestHashSource:
     """Tests for source hashing."""

tests/test_kida_modern_syntax.py

@@ -192,6 +192,22 @@ def test_no_precedence_warning_when_fallback_parenthesized(self, env):
             prec_warnings = [x for x in w if issubclass(x.category, PrecedenceWarning)]
             assert len(prec_warnings) == 1
 
+    def test_partial_eval_preserves_parenthesized_filter_fallback(self, env):
+        """Partial evaluation keeps the parser's parenthesized-filter marker."""
+        import warnings
+
+        from kida.exceptions import PrecedenceWarning
+
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            tmpl = env.from_string(
+                "{{ x ?? ((prefix ~ name) | upper) }}",
+                static_context={"prefix": "hi "},
+            )
+            assert tmpl.render(x=None, name="there") == "HI THERE"
+            prec_warnings = [x for x in w if issubclass(x.category, PrecedenceWarning)]
+            assert prec_warnings == []
+
     def test_null_coalesce_with_filter_requires_parens(self, env):
         """Filters must use parentheses to apply after null coalescing.