feat: enforce custom rule hermeticity at runtime

scripts/rules_benchmark/_helpers/workflow.py

@@ -39,6 +39,21 @@
 _CI_THRESHOLDS_PATH: Path = Path(__file__).parents[1] / "ci_thresholds.json"
 _GNU_TIME_PATH: Path = Path("/usr/bin/time")
 _PEAK_RSS_MARKER: str = "__SQLBUILD_BENCHMARK_PEAK_RSS_KIB__="
+_NON_CACHEABLE_PROBE_RULE: str = """
+import pathlib
+
+
+@rule(
+    code="XSQBRB999",
+    slug="non-cacheable-probe",
+    message="Non-cacheable probe observed the rules directory",
+    remediation="Remove the non-cacheable probe.",
+    enabled_by_default=True,
+)
+def non_cacheable_probe(*, model: Model, ctx: RuleContext) -> list[Finding]:
+    listed = pathlib.os.listdir(pathlib.Path(__file__).parent)
+    return [ctx.finding(subject=model)] * int(not listed)
+"""
 
 
 @dataclass(frozen=True)
@@ -890,7 +905,7 @@ def _extract_peak_rss(stderr: str) -> tuple[str, int | None]:
 def _measure_rejection(*, project_dir: Path, iterations: int) -> BenchmarkResult:
     rule_file: Path = project_dir / "rules" / "benchmark_rules.py"
     with rule_file.open("a", encoding="utf-8") as handle:
-        handle.write("\ndef non_cacheable_probe():\n    return open('untracked.txt')\n")
+        handle.write(_NON_CACHEABLE_PROBE_RULE)
     elapsed: list[float] = []
     peak_rss_bytes: list[int] = []
     for _ in range(iterations):

scripts/rules_benchmark/constants.py

@@ -8,5 +8,5 @@
 CI_STRESS_WARM_ITERATIONS: int = 3
 BENCHMARK_HELPER_RULE_INDEX: int = 14
 MAX_CUSTOM_RULE_COUNT: int = 999
-NON_CACHEABLE_REJECTION: str = "call to open must go through RuleContext"
+NON_CACHEABLE_REJECTION: str = "listing a directory is not allowed"
 SQL_SUFFIX: str = ".sql"

src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/rules/custom-rules/testing-and-determinism.md

@@ -31,20 +31,58 @@ def test_given_missing_order_id_when_evaluating_then_reports_finding() -> None:
 `RuleCase.files` adds supporting project files. `RuleCase.config` supplies option values. Include
 passing cases, failing cases, near misses, and exact source-position assertions.
 
+Each case also checks determinism. The harness evaluates every subject once, then evaluates every
+subject again in reverse order in the same process. If the Rule's findings differ between the two
+passes, the case fails with `custom rule <code> is not deterministic`. The usual cause is
+module-level state, such as a set of already-seen models, that carries over from one subject to
+the next. Compute findings from the subject and `ctx` alone.
+
 Repository pytest files remain outside the SQLBuild project's SQL `tests/` directory.
 
 ## Helpers
 
-Custom Rules can import repository-owned helpers under `rules/` and supported pure Python modules.
+Custom Rules can import repository-owned helpers under `rules/` and these deterministic standard
+library modules:
+
+`abc`, `bisect`, `collections`, `copy`, `dataclasses`, `decimal`, `difflib`, `enum`, `fnmatch`,
+`fractions`, `functools`, `graphlib`, `hashlib`, `heapq`, `itertools`, `json`, `math`,
+`operator`, `pathlib`, `re`, `statistics`, `string`, `textwrap`, `types`, `typing`, and
+`unicodedata`, plus `__future__` and `sqlbuild.rules`.
+
+Modules that expose the clock, randomness, the process, or the network are rejected, including
+`time`, `datetime`, `random`, `uuid`, `secrets`, `os`, `sys`, `subprocess`, `socket`, and
+`importlib`. SQLBuild checks the `import` statements in each selected Rule's file and in every
+helper under `rules/` it imports, directly or through other helpers, with absolute or relative
+imports anywhere in the module. These are the files the Rules cache fingerprints. Other Python files under
+`rules/`, such as tests or scratch scripts, are not checked.
+
 Changing an imported helper invalidates the Rules that depend on it.
 
 Only decorated functions register. Ordinary functions, constants, dataclasses, and classes remain
 helpers.
 
 ## Deterministic inputs
 
-Direct filesystem, environment, clock, randomness, network, and subprocess access is rejected.
-Those inputs cannot be reproduced safely by the Rules cache.
+The Rules cache can only reuse a result when the Rule depends on tracked inputs. SQLBuild runs
+custom Rules in a separate host process and enforces that at runtime:
+
+- Rules run with an empty environment, so `os.environ` is empty even when it is reached through
+  an allowed module, for example `pathlib.os.environ`.
+- The host starts with `PYTHONHASHSEED=0`, so iterating over a set of strings gives the same order
+  on every run.
+- The working directory is the resolved project directory, wherever `sqb` was invoked, so
+  `Path.cwd()` and relative paths do not vary between runs.
+- While Rule code runs, opening files, reading file metadata such as `Path.exists()` or
+  `Path.stat()`, listing directories, running commands, forking, network access, and `ctypes` are
+  rejected, however they are reached. Loading a module outside the
+  allowed list for the first time is rejected too.
+
+A rejected operation fails the command with `non-hermetic custom rule <code> at <file>:<line>`, even
+if the Rule catches the exception.
+
+The guard exists for cache correctness: it stops Rules from depending on inputs the cache cannot
+track by accident. It is not a security sandbox. Deliberately evasive code can bypass it, so only
+run Rules you trust.
 
 Read supported project text and structure through `ctx.project.tree`:
 
@@ -58,7 +96,8 @@ path invalidates the cached result.
 
 ## Cache granularity
 
-Cache identity incorporates the Rule implementation, imported helper closure, configured options,
-subject, accessed compiler facts, tracked project observations, dialect, and compatibility
-versions. Prefer model subjects for independent per-model checks and project subjects for genuine
+Cache identity incorporates the whole Rule source file, imported helper closure, configured
+options, subject, accessed compiler facts, tracked project observations, dialect, and
+compatibility versions. Editing any statement in a Rule file invalidates cached results for every
+Rule defined in that file, so keep unrelated Rules in separate files. Prefer model subjects for independent per-model checks and project subjects for genuine
 cross-project invariants.

src/sqlbuild/rule_engine/_helpers/engine/custom_rule_evidence.py

@@ -14,6 +14,7 @@
 _PUBLIC_HARNESS_MODULES: frozenset[str] = frozenset({"sqlbuild.rules", "sqlbuild.rules.testing"})
 _INIT_MODULE_NAME: str = "__init__"
 _RULES_DIRECTORY_NAME: str = "rules"
+_STAR_IMPORT_NAME: str = "*"
 _MIN_PARAMETRIZE_ARGUMENTS: int = 2
 _PARAMETRIZE_ATTRIBUTE: str = "parametrize"
 _PYTEST_MARK_ATTRIBUTE: str = "mark"
@@ -31,26 +32,6 @@ class _RuleTestEvidence:
     owner: str
 
 
-class _ModuleBindingVisitor(ast.NodeVisitor):
-    """Collect names bound in module control flow without entering local scopes."""
-
-    def __init__(self) -> None:
-        self.names: set[str] = set()
-
-    def visit_Name(self, node: ast.Name) -> None:
-        if isinstance(node.ctx, ast.Store):
-            self.names.add(node.id)
-
-    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
-        self.names.add(node.name)
-
-    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
-        self.names.add(node.name)
-
-    def visit_ClassDef(self, node: ast.ClassDef) -> None:
-        self.names.add(node.name)
-
-
 def custom_rule_test_evidence(*, rule: Rule, project_dir: Path) -> tuple[_RuleTestEvidence, ...]:
     """Return statically associated public-harness cases for one exact custom rule."""
 
@@ -87,66 +68,28 @@ def custom_rule_import_closure(*, rule: Rule, project_dir: Path) -> tuple[Path,
 def custom_rule_implementation_fingerprint(
     *, rule: Rule, project_dir: Path, import_closure: tuple[Path, ...] | None = None
 ) -> str:
-    """Fingerprint one rule function and its repository-owned imported helper closure."""
+    """Fingerprint one rule's whole source file and its repository-owned imported helper closure."""
 
     digest: Any = hashlib.sha256()
     if rule.source is None:
         digest.update(inspect.getsource(rule.check).encode())
         return digest.hexdigest()
-    digest.update(inspect.getsource(rule.check).encode())
-    digest.update(_module_binding_fingerprint(rule=rule))
     root: Path = project_dir.resolve()
+    source_path: Path = Path(rule.source).resolve()
     closure: tuple[Path, ...] = (
         custom_rule_import_closure(rule=rule, project_dir=root)
         if import_closure is None
         else import_closure
     )
-    for path in closure:
-        digest.update(path.relative_to(root).as_posix().encode())
+    for path in (source_path, *closure):
+        digest.update(_fingerprint_label(path=path, root=root).encode())
+        digest.update(b"\0")
         digest.update(path.read_bytes())
     return digest.hexdigest()
 
 
-def _module_binding_fingerprint(*, rule: Rule) -> bytes:
-    if rule.source is None:
-        return b""
-    source_path: Path = Path(rule.source).resolve()
-    try:
-        source: str = source_path.read_text(encoding="utf-8")
-        tree: ast.Module = ast.parse(source, filename=str(source_path))
-    except (OSError, UnicodeError, SyntaxError):
-        return b""
-    definitions: dict[str, ast.AST] = {}
-    for statement in tree.body:
-        visitor: _ModuleBindingVisitor = _ModuleBindingVisitor()
-        visitor.visit(statement)
-        for name in visitor.names:
-            definitions[name] = statement
-    check_name: str = getattr(rule.check, "__name__", "")
-    check_node: ast.AST | None = definitions.get(check_name)
-    pending: list[str] = [] if check_node is None else list(_loaded_names(check_node))
-    visited: set[str] = {check_name}
-    parts: list[bytes] = []
-    while pending:
-        name: str = pending.pop()
-        if name in visited:
-            continue
-        visited.add(name)
-        definition: ast.AST | None = definitions.get(name)
-        if definition is None:
-            continue
-        parts.append(name.encode())
-        parts.append(ast.dump(definition, include_attributes=False).encode())
-        pending.extend(_loaded_names(definition))
-    return b"\0".join(parts)
-
-
-def _loaded_names(node: ast.AST) -> tuple[str, ...]:
-    return tuple(
-        child.id
-        for child in ast.walk(node)
-        if isinstance(child, ast.Name) and isinstance(child.ctx, ast.Load)
-    )
+def _fingerprint_label(*, path: Path, root: Path) -> str:
+    return path.relative_to(root).as_posix() if path.is_relative_to(root) else path.as_posix()
 
 
 def custom_rule_project_fact_attributes(
@@ -182,62 +125,85 @@ def custom_rule_project_fact_attributes(
 
 
 def _import_closure(*, source_path: Path, project_dir: Path) -> tuple[Path, ...]:
+    rules_root: Path = (project_dir / _RULES_DIRECTORY_NAME).resolve()
     pending: list[Path] = [source_path]
     visited: set[Path] = {source_path}
     helpers: set[Path] = set()
     while pending:
         current: Path = pending.pop()
+        candidates: set[Path] = set(_package_inits(path=current, rules_root=rules_root))
         try:
             tree: ast.Module = ast.parse(current.read_text(encoding="utf-8"), filename=str(current))
         except (OSError, UnicodeError, SyntaxError):
-            continue
-        for node in tree.body:
-            candidate: Path | None = _local_import_path(
-                node=node,
-                source_path=current,
-                project_dir=project_dir,
+            tree = ast.Module(body=[], type_ignores=[])
+        for node in ast.walk(tree):
+            candidates.update(
+                _local_import_paths(node=node, source_path=current, rules_root=rules_root)
             )
-            if candidate is None or candidate in visited:
-                continue
+        for candidate in candidates - visited:
             visited.add(candidate)
             helpers.add(candidate)
             pending.append(candidate)
     return tuple(sorted(helpers))
 
 
-def _local_import_path(*, node: ast.stmt, source_path: Path, project_dir: Path) -> Path | None:
-    module: str | None = None
-    base: Path
-    level: int = 0
-    if isinstance(node, ast.ImportFrom):
-        module = node.module
-        level = node.level
+def _local_import_paths(*, node: ast.AST, source_path: Path, rules_root: Path) -> tuple[Path, ...]:
+    """Resolve every repository-owned rules module one import statement can execute."""
+
+    targets: list[tuple[Path, tuple[str, ...]]] = []
+    if isinstance(node, ast.Import):
+        targets.extend(
+            (rules_root.parent, tuple(alias.name.split(".")))
+            for alias in node.names
+            if alias.name.split(".")[0] == _RULES_DIRECTORY_NAME
+        )
+    elif isinstance(node, ast.ImportFrom):
+        base: Path = rules_root.parent
+        module_parts: tuple[str, ...] = tuple((node.module or "").split(".")) if node.module else ()
         if node.level:
             base = source_path.parent
             for _ in range(node.level - 1):
                 base = base.parent
-        else:
-            base = project_dir
-    elif isinstance(node, ast.Import) and len(node.names) == 1:
-        module = node.names[0].name
-        base = project_dir
-    else:
-        return None
-    if module is None:
-        return None
-    parts: tuple[str, ...] = tuple(module.split("."))
-    if not level and (not parts or parts[0] != _RULES_DIRECTORY_NAME):
-        return None
+        elif not module_parts or module_parts[0] != _RULES_DIRECTORY_NAME:
+            return ()
+        targets.append((base, module_parts))
+        targets.extend(
+            (base, (*module_parts, alias.name))
+            for alias in node.names
+            if alias.name != _STAR_IMPORT_NAME
+        )
+    resolved: list[Path] = []
+    for base, parts in targets:
+        module: Path | None = _module_file(base=base, parts=parts, rules_root=rules_root)
+        if module is not None:
+            resolved.append(module)
+            resolved.extend(_package_inits(path=module, rules_root=rules_root))
+    return tuple(resolved)
+
+
+def _module_file(*, base: Path, parts: tuple[str, ...], rules_root: Path) -> Path | None:
     candidate: Path = base.joinpath(*parts)
-    file_candidate: Path = candidate.with_suffix(".py")
-    package_candidate: Path = candidate / "__init__.py"
-    resolved: Path | None = file_candidate if file_candidate.is_file() else None
-    if resolved is None and package_candidate.is_file():
-        resolved = package_candidate
-    if resolved is None:
-        return None
-    resolved = resolved.resolve()
-    return resolved if resolved.is_relative_to(project_dir) else None
+    for path in (candidate / "__init__.py", candidate.with_suffix(".py") if parts else None):
+        if path is not None and path.is_file():
+            resolved: Path = path.resolve()
+            return resolved if resolved.is_relative_to(rules_root) else None
+    return None
+
+
+def _package_inits(*, path: Path, rules_root: Path) -> tuple[Path, ...]:
+    """Return package initializers executed before one module under the rules root."""
+
+    inits: list[Path] = []
+    for parent in path.parents:
+        if not parent.is_relative_to(rules_root):
+            break
+        init: Path = parent / "__init__.py"
+        if not init.is_file():
+            continue
+        resolved: Path = init.resolve()
+        if resolved != path and resolved.is_relative_to(rules_root):
+            inits.append(resolved)
+    return tuple(inits)
 
 
 def _file_evidence(