perf(test): route walk-tests through shared AST corpus

test/source_corpus.py

@@ -33,6 +33,7 @@
 
 import ast
 import functools
+import unicodedata
 from collections.abc import Iterator, Sequence
 from pathlib import Path
 
@@ -78,6 +79,30 @@ def unreadable_files() -> tuple[Path, ...]:
     return _read_tree()[1]
 
 
+def _nfkc(text: str) -> str:
+    """NFKC-normalise, matching how CPython folds identifiers at parse time."""
+    return unicodedata.normalize("NFKC", text)
+
+
+@functools.lru_cache(maxsize=1)
+def _normalized_texts() -> tuple[str, ...]:
+    """NFKC-normalised copy of every file's text, in ``source_texts`` order.
+
+    A gate matches on a bare identifier, and CPython NFKC-folds identifiers at
+    parse time -- so a call written with a Unicode compatibility homoglyph of a
+    guarded name (``delete_items_b\uff41tch``) is that ASCII name in the AST but
+    NOT in the raw bytes. Filtering on raw text would skip the file and let the
+    offender through green. Normalising the haystack (here) and the needle (in
+    ``candidate_sources``) the same way closes that hole while keeping the
+    narrowing: NFKC is a fixpoint on ASCII, so every raw ASCII match is
+    preserved and only homoglyph spellings are newly caught. Computed once over
+    the whole tree (~0.3s) and cached, like the read itself. ``source_texts``
+    still returns the RAW text, which gates that scan comments or string
+    literals (a ``# render-ok`` marker, an import alias) depend on.
+    """
+    return tuple(_nfkc(text) for _path, text in source_texts())
+
+
 def candidate_sources(
     require_all: Sequence[str] = (),
     require_any: Sequence[str] = (),
@@ -87,11 +112,18 @@ def candidate_sources(
     An empty ``require_any`` imposes no alternation, so passing neither argument
     returns the whole corpus.
     """
+    # Match on the NFKC-normalised text with NFKC-normalised needles, so a call
+    # whose identifier is a Unicode compatibility homoglyph of a literal (which
+    # CPython folds to that literal at parse time, making it a real AST match) is
+    # not skipped by a raw-byte pre-filter. The yielded ``text`` stays RAW.
+    all_n = tuple(_nfkc(lit) for lit in require_all)
+    any_n = tuple(_nfkc(lit) for lit in require_any)
+    texts = source_texts()
+    norm = _normalized_texts()
     return tuple(
         (path, text)
-        for path, text in source_texts()
-        if all(lit in text for lit in require_all)
-        and (not require_any or any(lit in text for lit in require_any))
+        for (path, text), ntext in zip(texts, norm)
+        if all(lit in ntext for lit in all_n) and (not any_n or any(lit in ntext for lit in any_n))
     )
 
 

test/test_knowledge_delete_off_loop.py

@@ -21,11 +21,11 @@
 import ast
 import asyncio
 import json
-import pathlib
 import threading
 from unittest.mock import MagicMock
 
 import pytest
+from source_corpus import parsed_candidates, src_root
 
 from kiro_crew.knowledge.folder_watcher import FolderWatcher
 from kiro_crew.knowledge.store import KnowledgeStore
@@ -35,7 +35,7 @@
 # repo's other on-loop guards use.
 _NESTED_SCOPES = (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)
 
-_SRC = pathlib.Path(__file__).resolve().parents[1] / "src" / "kiro_crew"
+_SRC = src_root()
 
 
 def _called_names(body: list[ast.stmt]) -> set[tuple[str, int]]:
@@ -101,11 +101,11 @@ def _on_loop_call_sites(name: str) -> list[str]:
     reaches it (see ``_sync_helpers_reaching``).
     """
     found: list[str] = []
-    for path in sorted(_SRC.rglob("*.py")):
-        try:
-            tree = ast.parse(path.read_text(errors="replace"))
-        except SyntaxError:  # pragma: no cover - syntax is enforced elsewhere
-            continue
+    # Only files whose TEXT holds ``name`` can call it (directly, or through a
+    # same-module sync helper that does), so the shared corpus parses just those
+    # instead of re-walking all ~1250 modules for this gate. ``src_root()`` is the
+    # same tree the old ``_SRC`` named, so the relative paths below are unchanged.
+    for path, _text, tree in parsed_candidates(require_all=(name,)):
         indirect = _sync_helpers_reaching(tree, name)
         for fn in (n for n in ast.walk(tree) if isinstance(n, ast.AsyncFunctionDef)):
             for called, lineno in sorted(_called_names(fn.body), key=lambda c: c[1]):

test/test_slack_render_pipeline.py

@@ -53,6 +53,7 @@
 import tokenize
 
 import pytest
+from source_corpus import parsed_candidates
 
 from kiro_crew.slack.format import (
     CONTINUATION,
@@ -174,22 +175,23 @@ def find_violations(source: str, path: str = "<source>") -> list[tuple[str, int,
 
 def collect_repo_violations() -> list[tuple[str, int, str]]:
     """Scan every ``kiro_crew/**/*.py`` except the owning module."""
-    root = _src_root()
-    base = root.parent
+    base = _src_root().parent
     out: list[tuple[str, int, str]] = []
-    for py in sorted(root.rglob("*.py")):
+    # A direct call to to_slack_mrkdwn -- bare-imported or reached as
+    # ``<module>.to_slack_mrkdwn`` -- can only exist in a file whose TEXT holds the
+    # literal ``to_slack_mrkdwn`` (the import that binds the alias, or the attribute
+    # call itself), so the shared corpus parses just those files rather than
+    # re-walking the whole package for this one gate. Narrowing on the literal
+    # drops non-matches only (proved lossless in the PR body).
+    for py, text, _tree in parsed_candidates(require_all=(_BANNED_FUNC,)):
         try:
             rel = str(py.relative_to(base))
         except ValueError:  # pragma: no cover - defensive
             rel = str(py)
         if rel.replace("\\", "/").endswith(_OWNER_MODULE):
             continue
         try:
-            src = py.read_text(encoding="utf-8")
-        except (OSError, UnicodeDecodeError):  # pragma: no cover - defensive
-            continue
-        try:
-            out.extend(find_violations(src, rel))
+            out.extend(find_violations(text, rel))
         except SyntaxError:  # pragma: no cover - defensive
             continue
     return out