fix(rules): keep split custom-rule hosts exact, bounded, and cancellable

Rules that change module-level state on any host are evaluated again in one
host over all of their subjects, replacing their split results, so cross-subject
state such as a duplicate detector reports exactly what one host reports. The
host count now honours cgroup v1 and v2 CPU quotas and is capped at eight.

The first failing host writes a per-run cancellation marker. The other hosts
check it between invocations, outside any Rule's guarded scope, and exit with a
distinct cancelled status, and hosts that start late exit at once; no process is
ever signalled. One invocation that is already running still ends or reaches its
own timeout. One host then reruns the plan to report the single-host error.

Author: kvlonge

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

@@ -6,6 +6,14 @@
 
 Online: https://sqlbuild.com/docs/concepts/rules/custom-rules/testing-and-determinism/
 
+## Contents
+
+- Test Rules
+- Parallel evaluation
+- Helpers
+- Deterministic inputs
+- Cache granularity
+
 ## Test Rules
 
 Use the public harness to exercise discovery, compilation, and Rules evaluation:
@@ -39,6 +47,19 @@ the next. Compute findings from the subject and `ctx` alone.
 
 Repository pytest files remain outside the SQLBuild project's SQL `tests/` directory.
 
+## Parallel evaluation
+
+On large projects, SQLBuild splits custom Rule evaluation across several host processes, at most
+one per available CPU and never more than eight. Each process evaluates a share of the subjects,
+and SQLBuild merges the findings back into the order a single process reports them.
+
+A Rule that changes module-level state while it runs, for example by adding each model to a set of
+already-seen names, could report different findings when its subjects are split. SQLBuild detects
+the state change and evaluates that Rule again in one process over all of its subjects, so its
+findings and cache entries match a single-process run. The result stays correct, but the Rule no
+longer runs in parallel and costs an extra host process. Keep each subject's findings independent
+of the others so Rules can run in parallel.
+
 ## Helpers
 
 Custom Rules can import repository-owned helpers under `rules/` and these deterministic standard

src/sqlbuild/rule_engine/_helpers/host/custom_evaluation.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from collections import Counter
+from collections.abc import Callable
 from contextlib import AbstractContextManager, nullcontext
 from dataclasses import dataclass
 from itertools import chain
@@ -22,13 +23,18 @@
     build_rule_fact_views,
 )
 from sqlbuild.rule_engine.classes.runtime_guard import RuntimeGuard
-from sqlbuild.rule_engine.constants import CUSTOM_RULE_PROJECT_SUBJECT
+from sqlbuild.rule_engine.constants import (
+    CUSTOM_HOST_CANCELLED_MESSAGE,
+    CUSTOM_RULE_PROJECT_SUBJECT,
+)
 from sqlbuild.rule_engine.exceptions import (
+    HostCancelledError,
     NonHermeticRuleError,
     OpaqueModuleStateError,
     RulesError,
 )
 from sqlbuild.rule_engine.models import (
+    CustomHostPartition,
     CustomRuleEvaluation,
     CustomRuleRun,
     Finding,
@@ -52,6 +58,7 @@ class _PassInputs:
     facts: RuleFactViews
     plan: CustomRulePlan | None
     guard: RuntimeGuard | None
+    cancelled: Callable[[], bool] | None
 
 
 def evaluate_custom_rules(
@@ -65,6 +72,7 @@ def evaluate_custom_rules(
     track_reads: bool = False,
     verify_determinism: bool = False,
     guard: RuntimeGuard | None = None,
+    partition: CustomHostPartition | None = None,
 ) -> CustomRuleRun:
     """Run selected custom rules for every planned subject through the Python authoring API."""
 
@@ -81,7 +89,10 @@ def evaluate_custom_rules(
     )
     reads: FactReads | None = FactReads() if track_reads else None
     tracker: _ReadTracker = _ReadTracker(
-        reads=reads, rules=custom_rules, rules_root=project_dir / "rules"
+        reads=reads,
+        rules=custom_rules,
+        rules_root=project_dir / "rules",
+        detect_state=partition is not None,
     )
 
     inputs: _PassInputs = _PassInputs(
@@ -93,6 +104,7 @@ def evaluate_custom_rules(
         facts=facts,
         plan=plan,
         guard=guard,
+        cancelled=None if partition is None else partition.cancelled,
     )
 
     def run_pass(
@@ -119,22 +131,29 @@ def run_pass(
         untracked_codes=frozenset(tracker.untracked_codes),
         uncacheable_codes=frozenset(tracker.uncacheable_codes),
         observed=() if reads is None else tuple(reads.observed.items()),
+        stateful_codes=frozenset(tracker.stateful_codes),
     )
 
 
 class _ReadTracker:
     """Attribute reads to one invocation; rules whose module state changes become untracked."""
 
     def __init__(
-        self, *, reads: FactReads | None, rules: tuple[Rule, ...], rules_root: Path
+        self,
+        *,
+        reads: FactReads | None,
+        rules: tuple[Rule, ...],
+        rules_root: Path,
+        detect_state: bool = False,
     ) -> None:
         self.reads: FactReads | None = reads
         self.untracked_codes: set[str] = set()
         self.uncacheable_codes: set[str] = set()
+        self.stateful_codes: set[str] = set()
         self._rules_root: Path = rules_root
         self._namespaces: dict[str, tuple[dict[str, object], ...]] = (
             {rule.code: rule_namespaces((rule.check,)) for rule in rules}
-            if reads is not None
+            if reads is not None or detect_state
             else {}
         )
         self._states: dict[str, str | None] = {
@@ -149,20 +168,22 @@ def start(self) -> None:
 
     def finish(self, *, code: str) -> frozenset[tuple[object, ...]] | None:
         reads: FactReads | None = self.reads
-        if reads is None:
-            return None
-        observed: set[tuple[object, ...]] | None = reads.current
-        reads.current = None
-        if reads.untracked:
-            self.untracked_codes.add(code)
-        if reads.uncacheable:
-            self.uncacheable_codes.add(code)
+        observed: set[tuple[object, ...]] | None = None
+        if reads is not None:
+            observed = reads.current
+            reads.current = None
+            if reads.untracked:
+                self.untracked_codes.add(code)
+            if reads.uncacheable:
+                self.uncacheable_codes.add(code)
         for rule_code, namespaces in self._namespaces.items():
-            if rule_code in self.untracked_codes:
+            if rule_code in self.stateful_codes:
                 continue
             state: str | None = self._state_token(namespaces)
             if state is None or state != self._states[rule_code]:
-                self.untracked_codes.add(rule_code)
+                self.stateful_codes.add(rule_code)
+                if reads is not None:
+                    self.untracked_codes.add(rule_code)
         return None if observed is None else frozenset(observed)
 
     def _state_token(self, namespaces: tuple[dict[str, object], ...]) -> str | None:
@@ -216,6 +237,7 @@ def context(rule: Rule) -> EvaluationRuleContext:
         if planned_project is not None and CUSTOM_RULE_PROJECT_SUBJECT not in planned_project:
             continue
         ctx: EvaluationRuleContext = context(rule)
+        _stop_if_cancelled(inputs)
         tracker.start()
         try:
             findings: list[Finding] = _invoke(
@@ -245,6 +267,7 @@ def context(rule: Rule) -> EvaluationRuleContext:
         for rule, ctx, planned in model_contexts:
             if planned is not None and path not in planned:
                 continue
+            _stop_if_cancelled(inputs)
             tracker.start()
             try:
                 findings = _invoke(rule=rule, subject=subject, ctx=ctx, guard=guard)
@@ -265,6 +288,13 @@ def context(rule: Rule) -> EvaluationRuleContext:
     return evaluations
 
 
+def _stop_if_cancelled(inputs: _PassInputs) -> None:
+    """Stop between invocations, outside any Rule's guarded scope, once the run is cancelled."""
+
+    if inputs.cancelled is not None and inputs.cancelled():
+        raise HostCancelledError(CUSTOM_HOST_CANCELLED_MESSAGE)
+
+
 def _invoke(
     *, rule: Rule, subject: object, ctx: EvaluationRuleContext, guard: RuntimeGuard | None
 ) -> list[Finding]:

src/sqlbuild/rule_engine/_helpers/host/custom_host.py

@@ -18,15 +18,23 @@
 from sqlbuild.rule_engine._helpers.host.custom_evaluation import evaluate_custom_rules
 from sqlbuild.rule_engine.classes.runtime_guard import RuntimeGuard
 from sqlbuild.rule_engine.constants import (
+    CUSTOM_HOST_CANCELLED_EXIT_CODE,
+    CUSTOM_HOST_CANCELLED_MESSAGE,
     CUSTOM_HOST_HASH_SEED,
     CUSTOM_HOST_INPUT_TUPLE_SIZE,
     CUSTOM_HOST_LAUNCH_MODULE,
     CUSTOM_HOST_MAX_TRACKED_READS,
     CUSTOM_HOST_PROTOCOL_VERSION,
     CUSTOM_HOST_RUNTIME_VERSION,
 )
-from sqlbuild.rule_engine.exceptions import RulesError
-from sqlbuild.rule_engine.models import CustomRuleRun, Finding, Rule, RulesConfig
+from sqlbuild.rule_engine.exceptions import HostCancelledError, RulesError
+from sqlbuild.rule_engine.models import (
+    CustomHostPartition,
+    CustomRuleRun,
+    Finding,
+    Rule,
+    RulesConfig,
+)
 from sqlbuild.rule_engine.types import CustomRulePlan
 
 
@@ -50,6 +58,9 @@ def main() -> int:
         payload: object = request["payload"]
         if not isinstance(payload, dict):
             raise RulesError("custom host payload must be an object")
+        cancel_marker: Path | None = _cancel_marker(payload)
+        if cancel_marker is not None and cancel_marker.exists():
+            raise HostCancelledError(CUSTOM_HOST_CANCELLED_MESSAGE)
         project, config = _decode_inputs(payload)
         project_dir: Path = Path(str(payload["project_dir"])).resolve()
         dialect: str = str(payload.get("dialect", "generic"))
@@ -77,7 +88,15 @@ def main() -> int:
                 track_reads=track_reads,
                 verify_determinism=verify_determinism,
                 guard=guard,
+                partition=(
+                    None
+                    if cancel_marker is None
+                    else CustomHostPartition(cancelled=cancel_marker.exists)
+                ),
             )
+    except HostCancelledError:
+        sys.stderr.write(CUSTOM_HOST_CANCELLED_MESSAGE)
+        return CUSTOM_HOST_CANCELLED_EXIT_CODE
     except Exception as error:
         violation: Exception | None = None if guard is None else guard.violation
         return _write_error(str(violation or error))
@@ -114,6 +133,20 @@ def _decode_inputs(payload: dict[str, Any]) -> tuple[CompiledProject, RulesConfi
     return decoded
 
 
+def _cancel_marker(payload: dict[str, Any]) -> Path | None:
+    """Return the run's cancellation marker, which must live beside the project payload."""
+
+    value: object = payload.get("cancel_marker")
+    if value is None:
+        return None
+    project_dir: Path = Path(str(payload["project_dir"])).resolve()
+    input_root: Path = (project_dir / "target" / "rules-cache" / "host-inputs").resolve()
+    marker: Path = Path(str(value))
+    if not marker.is_absolute() or not marker.resolve().is_relative_to(input_root):
+        raise RulesError("custom host cancellation marker path is invalid")
+    return marker
+
+
 def _decode_plan(value: object) -> CustomRulePlan:
     if not isinstance(value, dict):
         raise RulesError("custom host plan must be an object")
@@ -164,6 +197,7 @@ def _run_payload(run: CustomRuleRun) -> dict[str, object]:
         "uncacheable": sorted(run.uncacheable_codes),
         "observed": sorted([list(key), digest] for key, digest in run.observed),
         "bounded": bounded,
+        "stateful": sorted(run.stateful_codes),
     }