feat(policy): YAML-driven rule engine + parallel batch evaluator

Adds a small, declarative rule engine on top of the existing
PolicyProgram. Six ops only — lt, lte, eq, in, and, or — with an
AND/OR reduction across clauses. Anything beyond that is the job of
the general FHE evaluator, not a policy gate.

  rule_engine.go   parser + compiler (YAML -> RulePlan)
  rule_executor.go step dispatch + per-clause FHE primitive calls
  rule_batch.go    N-intent fan-out, one Engine per worker
  plan_cache.go    compile-once-per-(policy_id, sha256(yaml))

Five example wallet-tier policies (treasury_cold, hot_wallet_allowlist,
bridge_routing, validator_staking, gas_topup) ship under examples/
and are exercised by the compile suite.

Concurrency: luxfi/fhe.Evaluator is NOT goroutine-safe (internal
blind-rotation scratch). Batch executor therefore uses one Engine
per worker via EngineFactory; bootstrap key + parameters are
read-only and shared. Measured ~4x speedup on a 4-worker M1 batch
of 16 intents (33s -> 9s for the smoke plan).

Tests: 28 sub-cases across DSL parsing, malformed-input rejection,
example compile, AND/OR evaluation, missing-operand rejection,
batch evaluation at conc={1,4}, and plan-cache hit/forget/distinct-hash.
All pass against luxfi/fhe primitives at PN10QP27.

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/spf13/cobra v1.10.2
 	github.com/urfave/cli/v3 v3.6.2
 	golang.org/x/crypto v0.49.0
+	gopkg.in/yaml.v3 v3.0.1
 )
 
 // Local development against PolyDomain + NTTContext + ReductionBudget added
@@ -100,7 +101,6 @@ require (
 	google.golang.org/grpc v1.80.0 // indirect
 	google.golang.org/protobuf v1.36.11 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
-	gopkg.in/yaml.v3 v3.0.1 // indirect
 	modernc.org/libc v1.70.0 // indirect
 	modernc.org/mathutil v1.7.1 // indirect
 	modernc.org/memory v1.11.0 // indirect

policy/examples/bridge_routing.yaml

@@ -0,0 +1,30 @@
+rule_engine_version: "1"
+policy_id: "bridge-routing-v1"
+combinator: and
+clauses:
+  - name: amount_under_bridge_cap
+    op: lte
+    field: encryptedAmount
+    constant: encryptedBridgeCap
+    field_width: u64
+  - name: source_chain_allowed
+    op: in
+    field: encryptedSourceChain
+    constants:
+      - encryptedChainEthereum
+      - encryptedChainPolygon
+      - encryptedChainArbitrum
+      - encryptedChainOptimism
+    field_width: u32
+  - name: dest_chain_allowed
+    op: in
+    field: encryptedDestChain
+    constants:
+      - encryptedChainLuxC
+      - encryptedChainLuxX
+    field_width: u32
+  - name: per_window_cap
+    op: lt
+    field: encryptedSum
+    constant: encryptedWindowCap
+    field_width: u64

policy/examples/gas_topup.yaml

@@ -0,0 +1,22 @@
+rule_engine_version: "1"
+policy_id: "gas-topup-v1"
+combinator: and
+clauses:
+  - name: topup_under_per_call_cap
+    op: lt
+    field: encryptedAmount
+    constant: encryptedPerCallCap
+    field_width: u64
+  - name: target_is_known_operational_wallet
+    op: in
+    field: encryptedDestination
+    constants:
+      - encryptedOpsWallet1
+      - encryptedOpsWallet2
+      - encryptedOpsWallet3
+    field_width: u160
+  - name: window_topup_under_cap
+    op: lt
+    field: encryptedSum
+    constant: encryptedDailyTopupCap
+    field_width: u64

policy/examples/hot_wallet_allowlist.yaml

@@ -0,0 +1,22 @@
+rule_engine_version: "1"
+policy_id: "hot-wallet-v1"
+combinator: and
+clauses:
+  - name: amount_under_per_tx_cap
+    op: lt
+    field: encryptedAmount
+    constant: encryptedPerTxCap
+    field_width: u64
+  - name: dest_known
+    op: in
+    field: encryptedDestination
+    constants:
+      - encryptedAddr1
+      - encryptedAddr2
+      - encryptedAddr3
+      - encryptedAddr4
+      - encryptedAddr5
+      - encryptedAddr6
+      - encryptedAddr7
+      - encryptedAddr8
+    field_width: u160

policy/examples/treasury_cold.yaml

@@ -0,0 +1,22 @@
+rule_engine_version: "1"
+policy_id: "treasury-cold-v1"
+combinator: and
+clauses:
+  - name: amount_under_limit
+    op: lt
+    field: encryptedAmount
+    constant: encryptedAmountLimit
+    field_width: u64
+  - name: dest_in_allowlist
+    op: in
+    field: encryptedDestination
+    constants:
+      - encryptedAddrA
+      - encryptedAddrB
+      - encryptedAddrC
+    field_width: u160
+  - name: velocity_under_cap
+    op: lt
+    field: encryptedSum
+    constant: encryptedVelocityCap
+    field_width: u64

policy/examples/validator_staking.yaml

@@ -0,0 +1,26 @@
+rule_engine_version: "1"
+policy_id: "validator-staking-v1"
+combinator: and
+clauses:
+  - name: stake_amount_at_least_min
+    # Encoded as "amount > min" via lt with operands swapped at encrypt time:
+    # Min < Amount  ⇔  Amount > Min. We keep the DSL surface to lt/lte/eq/in.
+    op: lt
+    field: encryptedMinStake
+    constant: encryptedAmount
+    field_width: u64
+  - name: stake_amount_at_most_max
+    op: lte
+    field: encryptedAmount
+    constant: encryptedMaxStake
+    field_width: u64
+  - name: validator_in_active_set
+    op: in
+    field: encryptedValidatorID
+    constants:
+      - encryptedValidator1
+      - encryptedValidator2
+      - encryptedValidator3
+      - encryptedValidator4
+      - encryptedValidator5
+    field_width: u160

policy/plan_cache.go

@@ -0,0 +1,129 @@
+// Copyright (c) 2026, Lux Industries Inc
+// SPDX-License-Identifier: BSD-3-Clause
+
+// Plan cache: compile YAML once per policy version, reuse across every
+// subsequent evaluation. The cache is keyed by (PolicyID, sha256(YAML))
+// so any byte change in the policy document evicts the old entry —
+// there is no "soft" policy invalidation, no TTL, no stale eviction
+// after-the-fact. If the bytes match, the plan is reused; if not, the
+// caller compiles a new one.
+//
+// The cache is a thin wrapper around sync.Map: lock-free reads on the
+// hot path. Compile happens behind a singleflight-style lock so
+// concurrent miss-by-key callers do not redundantly parse the same
+// YAML.
+
+package policy
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+)
+
+// PlanLoader fetches the canonical YAML bytes for a given (policyID,
+// policyHash) from F-Chain PolicyVault. Callers — typically the
+// FHEVerifier in luxfi/mpc — supply this. The cache treats the loader
+// as the source of truth: if the loader returns bytes whose sha256
+// does not match the requested hash, GetOrCompile rejects the
+// response.
+type PlanLoader interface {
+	LoadYAML(policyID string, policyHash [32]byte) ([]byte, error)
+}
+
+// PlanLoaderFunc adapts a function to PlanLoader.
+type PlanLoaderFunc func(policyID string, policyHash [32]byte) ([]byte, error)
+
+// LoadYAML calls f.
+func (f PlanLoaderFunc) LoadYAML(policyID string, policyHash [32]byte) ([]byte, error) {
+	return f(policyID, policyHash)
+}
+
+// PlanCache memoizes compiled RulePlans by (policyID, policyHash).
+//
+// The zero value is ready to use. Callers may share one PlanCache
+// across all FHEVerifier instances in a process; entries never grow
+// large (a RulePlan is ~hundreds of bytes plus retained YAML).
+type PlanCache struct {
+	plans sync.Map // map[planKey]*RulePlan
+	mu    sync.Mutex
+}
+
+type planKey struct {
+	id   string
+	hash [32]byte
+}
+
+// ErrHashMismatch is returned by GetOrCompile when the loader returns
+// bytes that do not hash to the requested PolicyHash. F-Chain MUST be
+// authoritative on the bytes; serving a different policy than the
+// caller asked for is a bug, not a backwards-compat case.
+var ErrHashMismatch = errors.New("plan_cache: policy hash mismatch")
+
+// GetOrCompile returns a cached RulePlan for (policyID, policyHash) or
+// invokes loader to fetch the YAML bytes and compile a new one.
+//
+// Concurrent callers requesting the same key share the compile work:
+// the first caller compiles, subsequent callers wait on the same
+// mutex and see the cached result.
+func (c *PlanCache) GetOrCompile(policyID string, policyHash [32]byte, loader PlanLoader) (*RulePlan, error) {
+	if loader == nil {
+		return nil, fmt.Errorf("plan_cache: nil loader")
+	}
+	if policyID == "" {
+		return nil, fmt.Errorf("plan_cache: empty policy_id")
+	}
+
+	key := planKey{id: policyID, hash: policyHash}
+	if v, ok := c.plans.Load(key); ok {
+		return v.(*RulePlan), nil
+	}
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	// re-check inside the lock to avoid double-compile under contention
+	if v, ok := c.plans.Load(key); ok {
+		return v.(*RulePlan), nil
+	}
+
+	data, err := loader.LoadYAML(policyID, policyHash)
+	if err != nil {
+		return nil, fmt.Errorf("plan_cache: load %s: %w", policyID, err)
+	}
+	if got := HashYAML(data); got != policyHash {
+		return nil, fmt.Errorf("%w: id=%s want=%x got=%x", ErrHashMismatch, policyID, policyHash[:8], got[:8])
+	}
+
+	plan, err := Compile(data)
+	if err != nil {
+		return nil, fmt.Errorf("plan_cache: compile %s: %w", policyID, err)
+	}
+	if plan.PolicyID != policyID {
+		return nil, fmt.Errorf("plan_cache: policy_id mismatch in YAML: want=%s got=%s", policyID, plan.PolicyID)
+	}
+	c.plans.Store(key, plan)
+	return plan, nil
+}
+
+// Forget evicts every cached entry for policyID, regardless of hash. A
+// version bump on F-Chain that retires the prior bytes calls Forget so
+// the next evaluation reloads.
+func (c *PlanCache) Forget(policyID string) {
+	c.plans.Range(func(k, _ any) bool {
+		if k.(planKey).id == policyID {
+			c.plans.Delete(k)
+		}
+		return true
+	})
+}
+
+// Len returns the number of cached plans. Useful in tests to assert
+// the cache hit/miss behaviour.
+func (c *PlanCache) Len() int {
+	n := 0
+	c.plans.Range(func(_, _ any) bool {
+		n++
+		return true
+	})
+	return n
+}

policy/rule_batch.go

@@ -0,0 +1,155 @@
+// Copyright (c) 2026, Lux Industries Inc
+// SPDX-License-Identifier: BSD-3-Clause
+
+// Batch executor: parallel evaluation of one compiled RulePlan against
+// N intent-specific Operands.
+//
+// The natural parallel-signing pattern is N transactions arriving in a
+// single round; each one needs the same policy evaluated against its
+// own encrypted inputs. This file evaluates them concurrently across
+// the available CPU cores using a worker pool. The same entry point is
+// the natural place to wire GPU dispatch later — when the FHE stack
+// exposes a path that batches N evaluations in one kernel launch, this
+// function is what swaps to it.
+//
+// Concurrency note: luxfi/fhe.Evaluator (and BitwiseEvaluator) hold
+// internal scratch buffers used by blind-rotation. Sharing one
+// Evaluator across goroutines is NOT safe — concurrent evaluators
+// will race on those buffers and panic. Each worker therefore owns
+// its own Engine constructed via the supplied EngineFactory. The
+// BootstrapKey + Parameters they share are read-only and may be
+// reused freely.
+
+package policy
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"runtime"
+	"sync"
+	"time"
+
+	"github.com/luxfi/fhe"
+)
+
+// EngineFactory builds a fresh Engine for one worker. Implementations
+// typically close over a (Parameters, *BootstrapKey) pair.
+type EngineFactory func() *Engine
+
+// NewEngineFactory returns the canonical EngineFactory for a given
+// FHE parameter set + bootstrap key. The returned factory allocates
+// a fresh BitwiseEvaluator + Evaluator on each call; the caller is
+// responsible for not invoking it more times than needed.
+func NewEngineFactory(params fhe.Parameters, bsk *fhe.BootstrapKey) EngineFactory {
+	return func() *Engine { return NewEngine(params, bsk) }
+}
+
+// BatchRequest is N transactions sharing one policy. Operands[i] is the
+// encrypted input set for transaction i. Constants live alongside the
+// request; the batch executor does not duplicate them.
+type BatchRequest struct {
+	// Inputs holds N per-intent encrypted input maps. len(Inputs) == N.
+	Inputs []map[string]*fhe.BitCiphertext
+
+	// Constants are the per-policy ciphertexts shared across all N
+	// evaluations. They are the same pointers passed to single-shot
+	// Evaluate; no copy is made. Ciphertexts are read-only during
+	// evaluation, so sharing across workers is safe.
+	Constants map[string]*fhe.BitCiphertext
+
+	// Concurrency caps the number of worker goroutines (each owning
+	// one Engine). Zero or negative means runtime.NumCPU(). The cap
+	// is honoured exactly; overflow is queued and dispatched as
+	// workers free.
+	Concurrency int
+}
+
+// BatchResult carries the encrypted verdicts and timing metadata.
+type BatchResult struct {
+	// Verdicts[i] is the encrypted boolean produced by Inputs[i]. On
+	// any per-intent failure the entire batch returns an error; partial
+	// results are not surfaced.
+	Verdicts []*fhe.Ciphertext
+
+	// Latency is wall-clock time spent inside EvaluateBatch.
+	Latency time.Duration
+}
+
+// EvaluateBatch fans N intents across a worker pool and returns N
+// encrypted verdicts. Each worker owns its own Engine constructed via
+// factory — required because luxfi/fhe.Evaluator is not goroutine-safe.
+//
+// Failures fail the whole batch — a stuck or noisy intent is treated
+// as a deny signal upstream, which is the safe posture for a signing
+// gate.
+func (p *RulePlan) EvaluateBatch(ctx context.Context, factory EngineFactory, batch BatchRequest) (BatchResult, error) {
+	if factory == nil {
+		return BatchResult{}, errors.New("rule_executor: nil engine factory")
+	}
+	n := len(batch.Inputs)
+	if n == 0 {
+		return BatchResult{}, errors.New("rule_executor: empty batch")
+	}
+
+	workers := batch.Concurrency
+	if workers <= 0 {
+		workers = runtime.NumCPU()
+	}
+	if workers > n {
+		workers = n
+	}
+
+	verdicts := make([]*fhe.Ciphertext, n)
+	jobs := make(chan int, n)
+	for i := 0; i < n; i++ {
+		jobs <- i
+	}
+	close(jobs)
+
+	var wg sync.WaitGroup
+	errCh := make(chan error, workers)
+
+	cctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	start := time.Now()
+
+	for w := 0; w < workers; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			eng := factory()
+			for i := range jobs {
+				if cctx.Err() != nil {
+					return
+				}
+				v, err := p.Evaluate(cctx, eng, Operands{
+					Inputs:    batch.Inputs[i],
+					Constants: batch.Constants,
+				})
+				if err != nil {
+					select {
+					case errCh <- fmt.Errorf("intent[%d]: %w", i, err):
+						cancel()
+					default:
+					}
+					return
+				}
+				verdicts[i] = v
+			}
+		}()
+	}
+
+	wg.Wait()
+	close(errCh)
+
+	if err, ok := <-errCh; ok {
+		return BatchResult{}, err
+	}
+
+	return BatchResult{
+		Verdicts: verdicts,
+		Latency:  time.Since(start),
+	}, nil
+}

policy/rule_engine.go

@@ -0,0 +1,244 @@
+// Copyright (c) 2026, Lux Industries Inc
+// SPDX-License-Identifier: BSD-3-Clause
+
+// Rule engine: a small declarative DSL that compiles a YAML policy
+// document to an executable RulePlan against luxfi/fhe primitives.
+//
+// The DSL is intentionally narrow. It expresses exactly the ops F-Chain
+// PolicyVault clauses need today — strict-less-than, less-than-or-equal,
+// equality, set membership, and an AND/OR reduction across clauses.
+// Arithmetic, multiplication, scalar fold, and signed integers are
+// out of scope; they are the job of the general FHE evaluator, not a
+// policy gate.
+//
+// Compilation is one-shot per (policy_id, policy_hash) and is cached in
+// plan_cache.go. The compiled plan is read-only and safe to share across
+// goroutines; per-evaluation state lives entirely in the input map and
+// the per-step intermediates slice.
+
+package policy
+
+import (
+	"crypto/sha256"
+	"errors"
+	"fmt"
+
+	"github.com/luxfi/fhe"
+	"gopkg.in/yaml.v3"
+)
+
+// SchemaVersion is the YAML schema version this engine accepts. Bumping
+// the schema version requires a new field in YAMLDocument, not an
+// override of an existing field.
+const SchemaVersion = "1"
+
+// Op is the small enumerated set of operations a clause may use. Adding
+// a new op is a deliberate change: it requires a Step.run case + a
+// validate-time entry in opArity.
+type Op string
+
+const (
+	OpLt  Op = "lt"  // field < constant
+	OpLte Op = "lte" // field <= constant
+	OpEq  Op = "eq"  // field == constant
+	OpIn  Op = "in"  // field ∈ {constants...}
+)
+
+// Combinator selects how clause verdicts reduce to one verdict.
+type Combinator string
+
+const (
+	CombAnd Combinator = "and" // all clauses must pass
+	CombOr  Combinator = "or"  // any clause may pass
+)
+
+// FieldWidth declares the bit-width of the integer-valued field/constant
+// operands in a clause. The width must match the BitCiphertext.Type() of
+// the encrypted operand at runtime; mismatch is a configuration bug and
+// is rejected at evaluation time by luxfi/fhe.
+type FieldWidth string
+
+const (
+	WidthU4   FieldWidth = "u4"
+	WidthU8   FieldWidth = "u8"
+	WidthU16  FieldWidth = "u16"
+	WidthU32  FieldWidth = "u32"
+	WidthU64  FieldWidth = "u64"
+	WidthU128 FieldWidth = "u128"
+	WidthU160 FieldWidth = "u160" // address-shaped fields
+	WidthU256 FieldWidth = "u256"
+)
+
+// FheType maps a declared FieldWidth to the matching luxfi/fhe type. A
+// blank or unknown width returns 0 / FheBool which is rejected when the
+// operand width is checked at runtime.
+func (w FieldWidth) FheType() fhe.FheUintType {
+	switch w {
+	case WidthU4:
+		return fhe.FheUint4
+	case WidthU8:
+		return fhe.FheUint8
+	case WidthU16:
+		return fhe.FheUint16
+	case WidthU32:
+		return fhe.FheUint32
+	case WidthU64:
+		return fhe.FheUint64
+	case WidthU128:
+		return fhe.FheUint128
+	case WidthU160:
+		return fhe.FheUint160
+	case WidthU256:
+		return fhe.FheUint256
+	}
+	return fhe.FheBool
+}
+
+// YAMLClause is the on-disk shape of a single clause. Authors hand-edit
+// these or generate them from a higher-level policy compiler.
+type YAMLClause struct {
+	Name       string     `yaml:"name"`
+	Op         Op         `yaml:"op"`
+	Field      string     `yaml:"field"`
+	Constant   string     `yaml:"constant,omitempty"`
+	Constants  []string   `yaml:"constants,omitempty"`
+	FieldWidth FieldWidth `yaml:"field_width"`
+}
+
+// YAMLDocument is the top-level policy document.
+type YAMLDocument struct {
+	SchemaVersion string       `yaml:"rule_engine_version"`
+	PolicyID      string       `yaml:"policy_id"`
+	Clauses       []YAMLClause `yaml:"clauses"`
+	Combinator    Combinator   `yaml:"combinator"`
+}
+
+// EvaluationStep is one compiled clause. The executor walks Steps in
+// order; each Step produces a single encrypted boolean. The terminal
+// reduction (AND/OR across all Step results) is performed by the
+// executor — Steps never reduce against each other.
+//
+// EvaluationStep is read-only after Compile; do not mutate fields from
+// callers.
+type EvaluationStep struct {
+	Name       string
+	Op         Op
+	FieldKey   string   // input map key
+	ConstKey   string   // constant map key (lt/lte/eq) — empty for "in"
+	ConstKeys  []string // constant map keys (in)
+	FieldWidth FieldWidth
+}
+
+// RulePlan is a compiled, read-only policy program.
+type RulePlan struct {
+	Version    string
+	PolicyID   string
+	PolicyHash [32]byte
+	Steps      []EvaluationStep
+	Combinator Combinator
+
+	// rawYAML is retained so the plan can be re-hashed deterministically
+	// or re-emitted for audit trails. It is not used during evaluation.
+	rawYAML []byte
+}
+
+// HashYAML returns the canonical sha256 of the YAML bytes used to compile
+// the plan. This is the same hash the cache keys on.
+func HashYAML(data []byte) [32]byte {
+	return sha256.Sum256(data)
+}
+
+// Compile parses YAML and returns a ready-to-evaluate RulePlan. The
+// PolicyHash is the sha256 of the input bytes — callers MUST pass the
+// same canonical bytes used by F-Chain so cache keys agree.
+//
+// Compile is deterministic: equivalent inputs produce equivalent plans.
+// Failures are returned as errors; this never panics.
+func Compile(data []byte) (*RulePlan, error) {
+	if len(data) == 0 {
+		return nil, errors.New("rule_engine: empty policy document")
+	}
+	var doc YAMLDocument
+	if err := yaml.Unmarshal(data, &doc); err != nil {
+		return nil, fmt.Errorf("rule_engine: yaml: %w", err)
+	}
+	if err := validate(&doc); err != nil {
+		return nil, err
+	}
+	steps := make([]EvaluationStep, len(doc.Clauses))
+	for i, c := range doc.Clauses {
+		steps[i] = EvaluationStep{
+			Name:       c.Name,
+			Op:         c.Op,
+			FieldKey:   c.Field,
+			ConstKey:   c.Constant,
+			ConstKeys:  append([]string(nil), c.Constants...),
+			FieldWidth: c.FieldWidth,
+		}
+	}
+	return &RulePlan{
+		Version:    doc.SchemaVersion,
+		PolicyID:   doc.PolicyID,
+		PolicyHash: HashYAML(data),
+		Steps:      steps,
+		Combinator: doc.Combinator,
+		rawYAML:    append([]byte(nil), data...),
+	}, nil
+}
+
+// validate enforces structural invariants the executor relies on.
+//
+//   - schema version equals SchemaVersion
+//   - policy_id non-empty
+//   - clauses non-empty
+//   - each clause has a recognised op + matching operand shape
+//   - combinator is "and" or "or"
+//
+// Reaching the executor requires passing validate; the executor does
+// not re-check these invariants on the hot path.
+func validate(doc *YAMLDocument) error {
+	if doc.SchemaVersion != SchemaVersion {
+		return fmt.Errorf("rule_engine: unsupported schema version %q (want %q)", doc.SchemaVersion, SchemaVersion)
+	}
+	if doc.PolicyID == "" {
+		return errors.New("rule_engine: policy_id required")
+	}
+	if len(doc.Clauses) == 0 {
+		return errors.New("rule_engine: at least one clause required")
+	}
+	switch doc.Combinator {
+	case CombAnd, CombOr:
+	default:
+		return fmt.Errorf("rule_engine: unknown combinator %q", doc.Combinator)
+	}
+	for i, c := range doc.Clauses {
+		if c.Name == "" {
+			return fmt.Errorf("rule_engine: clause[%d]: name required", i)
+		}
+		if c.Field == "" {
+			return fmt.Errorf("rule_engine: clause[%d] %q: field required", i, c.Name)
+		}
+		if c.FieldWidth.FheType() == fhe.FheBool {
+			return fmt.Errorf("rule_engine: clause[%d] %q: unknown field_width %q", i, c.Name, c.FieldWidth)
+		}
+		switch c.Op {
+		case OpLt, OpLte, OpEq:
+			if c.Constant == "" {
+				return fmt.Errorf("rule_engine: clause[%d] %q: constant required for op %q", i, c.Name, c.Op)
+			}
+			if len(c.Constants) != 0 {
+				return fmt.Errorf("rule_engine: clause[%d] %q: constants forbidden for op %q (use constant)", i, c.Name, c.Op)
+			}
+		case OpIn:
+			if c.Constant != "" {
+				return fmt.Errorf("rule_engine: clause[%d] %q: constant forbidden for op in (use constants)", i, c.Name)
+			}
+			if len(c.Constants) == 0 {
+				return fmt.Errorf("rule_engine: clause[%d] %q: at least one constants entry required for op in", i, c.Name)
+			}
+		default:
+			return fmt.Errorf("rule_engine: clause[%d] %q: unknown op %q", i, c.Name, c.Op)
+		}
+	}
+	return nil
+}

policy/rule_engine_test.go

@@ -0,0 +1,581 @@
+// Copyright (c) 2026, Lux Industries Inc
+// SPDX-License-Identifier: BSD-3-Clause
+
+package policy
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync/atomic"
+	"testing"
+
+	"github.com/luxfi/fhe"
+)
+
+// --- DSL parsing -----------------------------------------------------
+
+func TestRuleEngine_ParseValid(t *testing.T) {
+	yamlBytes := []byte(`
+rule_engine_version: "1"
+policy_id: "treasury-cold-v4"
+combinator: and
+clauses:
+  - name: amount_under_limit
+    op: lt
+    field: encryptedAmount
+    constant: encryptedLimit
+    field_width: u4
+  - name: dest_in_allowlist
+    op: in
+    field: encryptedDestination
+    constants:
+      - encryptedAddr1
+      - encryptedAddr2
+      - encryptedAddr3
+    field_width: u4
+  - name: velocity_under_cap
+    op: lt
+    field: encryptedSum
+    constant: encryptedVelocityCap
+    field_width: u4
+`)
+	plan, err := Compile(yamlBytes)
+	if err != nil {
+		t.Fatalf("Compile: %v", err)
+	}
+	if plan.Version != "1" {
+		t.Errorf("version: got %q want 1", plan.Version)
+	}
+	if plan.PolicyID != "treasury-cold-v4" {
+		t.Errorf("policy_id: got %q", plan.PolicyID)
+	}
+	if plan.Combinator != CombAnd {
+		t.Errorf("combinator: got %q want and", plan.Combinator)
+	}
+	if len(plan.Steps) != 3 {
+		t.Fatalf("steps: got %d want 3", len(plan.Steps))
+	}
+	if plan.Steps[1].Op != OpIn {
+		t.Errorf("step 1 op: got %q want in", plan.Steps[1].Op)
+	}
+	if len(plan.Steps[1].ConstKeys) != 3 {
+		t.Errorf("step 1 const keys: got %d want 3", len(plan.Steps[1].ConstKeys))
+	}
+}
+
+func TestRuleEngine_ParseRejectsMalformed(t *testing.T) {
+	cases := []struct {
+		name string
+		body string
+		want string
+	}{
+		{
+			"empty",
+			``,
+			"empty policy document",
+		},
+		{
+			"wrong_version",
+			`rule_engine_version: "2"
+policy_id: x
+combinator: and
+clauses:
+  - {name: a, op: lt, field: f, constant: c, field_width: u4}`,
+			"unsupported schema version",
+		},
+		{
+			"missing_policy_id",
+			`rule_engine_version: "1"
+combinator: and
+clauses:
+  - {name: a, op: lt, field: f, constant: c, field_width: u4}`,
+			"policy_id required",
+		},
+		{
+			"no_clauses",
+			`rule_engine_version: "1"
+policy_id: x
+combinator: and
+clauses: []`,
+			"at least one clause required",
+		},
+		{
+			"unknown_combinator",
+			`rule_engine_version: "1"
+policy_id: x
+combinator: maybe
+clauses:
+  - {name: a, op: lt, field: f, constant: c, field_width: u4}`,
+			"unknown combinator",
+		},
+		{
+			"unknown_op",
+			`rule_engine_version: "1"
+policy_id: x
+combinator: and
+clauses:
+  - {name: a, op: gt, field: f, constant: c, field_width: u4}`,
+			"unknown op",
+		},
+		{
+			"in_with_constant",
+			`rule_engine_version: "1"
+policy_id: x
+combinator: and
+clauses:
+  - {name: a, op: in, field: f, constant: c, constants: [c1], field_width: u4}`,
+			"constant forbidden for op in",
+		},
+		{
+			"lt_without_constant",
+			`rule_engine_version: "1"
+policy_id: x
+combinator: and
+clauses:
+  - {name: a, op: lt, field: f, field_width: u4}`,
+			"constant required for op",
+		},
+		{
+			"in_without_constants",
+			`rule_engine_version: "1"
+policy_id: x
+combinator: and
+clauses:
+  - {name: a, op: in, field: f, field_width: u4}`,
+			"at least one constants entry required",
+		},
+		{
+			"bad_field_width",
+			`rule_engine_version: "1"
+policy_id: x
+combinator: and
+clauses:
+  - {name: a, op: lt, field: f, constant: c, field_width: u3}`,
+			"unknown field_width",
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := Compile([]byte(tc.body))
+			if err == nil {
+				t.Fatal("expected error, got nil")
+			}
+			if !strings.Contains(err.Error(), tc.want) {
+				t.Fatalf("error = %v, want contains %q", err, tc.want)
+			}
+		})
+	}
+}
+
+// --- Example YAML files compile -------------------------------------
+
+func TestRuleEngine_ExamplesCompile(t *testing.T) {
+	cases := []struct {
+		name      string
+		path      string
+		stepCount int
+	}{
+		{"treasury_cold", "examples/treasury_cold.yaml", 3},
+		{"hot_wallet_allowlist", "examples/hot_wallet_allowlist.yaml", 2},
+		{"bridge_routing", "examples/bridge_routing.yaml", 4},
+		{"validator_staking", "examples/validator_staking.yaml", 3},
+		{"gas_topup", "examples/gas_topup.yaml", 3},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			abs, err := filepath.Abs(tc.path)
+			if err != nil {
+				t.Fatalf("abs: %v", err)
+			}
+			data, err := os.ReadFile(abs)
+			if err != nil {
+				t.Fatalf("read %s: %v", abs, err)
+			}
+			plan, err := Compile(data)
+			if err != nil {
+				t.Fatalf("compile %s: %v", tc.name, err)
+			}
+			if len(plan.Steps) != tc.stepCount {
+				t.Fatalf("%s: steps got=%d want=%d", tc.name, len(plan.Steps), tc.stepCount)
+			}
+			// Hash is deterministic.
+			if plan.PolicyHash == [32]byte{} {
+				t.Fatal("policy hash is zero")
+			}
+		})
+	}
+}
+
+// --- Evaluation against luxfi/fhe primitives ------------------------
+
+// ruleCtx bundles FHE keys + 4-bit encrypt helpers — shared across the
+// evaluation tests so each test case stays terse.
+type ruleCtx struct {
+	params fhe.Parameters
+	bsk    *fhe.BootstrapKey
+	enc    *fhe.BitwiseEncryptor
+	dec    *fhe.Decryptor
+	eng    *Engine
+}
+
+func newRuleCtx(t testing.TB) *ruleCtx {
+	t.Helper()
+	params, err := fhe.NewParametersFromLiteral(fhe.PN10QP27)
+	if err != nil {
+		t.Fatalf("params: %v", err)
+	}
+	kg := fhe.NewKeyGenerator(params)
+	sk := kg.GenSecretKey()
+	bsk := kg.GenBootstrapKey(sk)
+	return &ruleCtx{
+		params: params,
+		bsk:    bsk,
+		enc:    fhe.NewBitwiseEncryptor(params, sk),
+		dec:    fhe.NewDecryptor(params, sk),
+		eng:    NewEngine(params, bsk),
+	}
+}
+
+func (c *ruleCtx) encU4(v uint8) *fhe.BitCiphertext {
+	return c.enc.EncryptUint64(uint64(v&0xF), fhe.FheUint4)
+}
+
+// inlineYAML mirrors the treasury-cold example but at u4 width so the
+// FHE evaluation completes inside the noise budget of PN10QP27.
+const inlineYAMLU4 = `
+rule_engine_version: "1"
+policy_id: "treasury-cold-test"
+combinator: and
+clauses:
+  - name: amount_under_limit
+    op: lt
+    field: encryptedAmount
+    constant: encryptedLimit
+    field_width: u4
+  - name: dest_in_allowlist
+    op: in
+    field: encryptedDestination
+    constants:
+      - encryptedAddr1
+      - encryptedAddr2
+      - encryptedAddr3
+    field_width: u4
+  - name: velocity_under_cap
+    op: lt
+    field: encryptedSum
+    constant: encryptedVelocityCap
+    field_width: u4
+`
+
+func TestRuleEngine_EvaluateMatchesPlaintext(t *testing.T) {
+	if testing.Short() {
+		t.Skip("FHE evaluation is slow; -short skips")
+	}
+	ctx := newRuleCtx(t)
+	plan, err := Compile([]byte(inlineYAMLU4))
+	if err != nil {
+		t.Fatalf("compile: %v", err)
+	}
+
+	constants := map[string]*fhe.BitCiphertext{
+		"encryptedLimit":       ctx.encU4(8),
+		"encryptedAddr1":       ctx.encU4(0x1),
+		"encryptedAddr2":       ctx.encU4(0x2),
+		"encryptedAddr3":       ctx.encU4(0x3),
+		"encryptedVelocityCap": ctx.encU4(8),
+	}
+
+	cases := []struct {
+		name   string
+		amount uint8
+		dest   uint8
+		used   uint8
+		want   bool
+	}{
+		{"all_pass", 2, 0x2, 2, true},
+		{"amount_over", 12, 0x2, 2, false},
+		{"velocity_over", 2, 0x2, 12, false},
+		{"dest_not_in_set", 2, 0xF, 2, false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			inputs := map[string]*fhe.BitCiphertext{
+				"encryptedAmount":      ctx.encU4(tc.amount),
+				"encryptedDestination": ctx.encU4(tc.dest),
+				"encryptedSum":         ctx.encU4(tc.used),
+			}
+			result, err := plan.Evaluate(context.Background(), ctx.eng, Operands{
+				Inputs:    inputs,
+				Constants: constants,
+			})
+			if err != nil {
+				t.Fatalf("evaluate: %v", err)
+			}
+			got := ctx.dec.Decrypt(result)
+			if got != tc.want {
+				t.Fatalf("amount=%d dest=0x%X used=%d: got %v, want %v",
+					tc.amount, tc.dest, tc.used, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestRuleEngine_EvaluateOrCombinator(t *testing.T) {
+	if testing.Short() {
+		t.Skip("FHE evaluation is slow; -short skips")
+	}
+	ctx := newRuleCtx(t)
+	plan, err := Compile([]byte(`
+rule_engine_version: "1"
+policy_id: "any-pass-v1"
+combinator: or
+clauses:
+  - name: a_below
+    op: lt
+    field: encryptedA
+    constant: encryptedThresh
+    field_width: u4
+  - name: b_eq_target
+    op: eq
+    field: encryptedB
+    constant: encryptedTarget
+    field_width: u4
+`))
+	if err != nil {
+		t.Fatalf("compile: %v", err)
+	}
+	constants := map[string]*fhe.BitCiphertext{
+		"encryptedThresh": ctx.encU4(8),
+		"encryptedTarget": ctx.encU4(0x5),
+	}
+
+	cases := []struct {
+		name string
+		a, b uint8
+		want bool
+	}{
+		{"only_a_passes", 2, 0xA, true}, // a < 8 ✓, b != 5
+		{"only_b_passes", 12, 0x5, true}, // a >= 8, b == 5 ✓
+		{"both_pass", 2, 0x5, true},
+		{"neither", 12, 0xA, false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			inputs := map[string]*fhe.BitCiphertext{
+				"encryptedA": ctx.encU4(tc.a),
+				"encryptedB": ctx.encU4(tc.b),
+			}
+			r, err := plan.Evaluate(context.Background(), ctx.eng, Operands{Inputs: inputs, Constants: constants})
+			if err != nil {
+				t.Fatalf("evaluate: %v", err)
+			}
+			if got := ctx.dec.Decrypt(r); got != tc.want {
+				t.Fatalf("a=%d b=0x%X: got %v want %v", tc.a, tc.b, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestRuleEngine_MissingOperandRejected(t *testing.T) {
+	ctx := newRuleCtx(t)
+	plan, err := Compile([]byte(inlineYAMLU4))
+	if err != nil {
+		t.Fatalf("compile: %v", err)
+	}
+	// Missing encryptedDestination.
+	inputs := map[string]*fhe.BitCiphertext{
+		"encryptedAmount": ctx.encU4(1),
+		"encryptedSum":    ctx.encU4(1),
+	}
+	constants := map[string]*fhe.BitCiphertext{
+		"encryptedLimit":       ctx.encU4(8),
+		"encryptedAddr1":       ctx.encU4(0x1),
+		"encryptedAddr2":       ctx.encU4(0x2),
+		"encryptedAddr3":       ctx.encU4(0x3),
+		"encryptedVelocityCap": ctx.encU4(8),
+	}
+	_, err = plan.Evaluate(context.Background(), ctx.eng, Operands{Inputs: inputs, Constants: constants})
+	if err == nil {
+		t.Fatal("expected ErrMissingOperand, got nil")
+	}
+}
+
+// --- Batch executor --------------------------------------------------
+
+func TestRuleEngine_EvaluateBatch(t *testing.T) {
+	if testing.Short() {
+		t.Skip("FHE evaluation is slow; -short skips")
+	}
+	ctx := newRuleCtx(t)
+	// Batch test uses a smaller 2-clause plan to keep gate depth well
+	// inside PN10QP27's noise budget over N=16 intents. Single-shot
+	// 3-clause evaluation is covered by TestRuleEngine_EvaluateMatchesPlaintext.
+	plan, err := Compile([]byte(`
+rule_engine_version: "1"
+policy_id: "batch-smoke-v1"
+combinator: and
+clauses:
+  - name: amount_under_limit
+    op: lt
+    field: encryptedAmount
+    constant: encryptedLimit
+    field_width: u4
+  - name: dest_eq_target
+    op: eq
+    field: encryptedDestination
+    constant: encryptedTarget
+    field_width: u4
+`))
+	if err != nil {
+		t.Fatalf("compile: %v", err)
+	}
+
+	constants := map[string]*fhe.BitCiphertext{
+		"encryptedLimit":  ctx.encU4(8),
+		"encryptedTarget": ctx.encU4(0x2),
+	}
+
+	// Batch of N alternating pass/deny intents. Each intent has its
+	// own freshly-encrypted ciphertexts so per-intent noise is
+	// independent. Values are chosen well clear of the limit (pass
+	// amount=1, deny amount=15) for clean noise margin.
+	const N = 16
+	inputs := make([]map[string]*fhe.BitCiphertext, N)
+	want := make([]bool, N)
+	for i := 0; i < N; i++ {
+		amt := uint8(1)
+		dest := uint8(0x2)
+		expected := true
+		if i%2 == 1 {
+			amt = 15
+			expected = false
+		}
+		inputs[i] = map[string]*fhe.BitCiphertext{
+			"encryptedAmount":      ctx.encU4(amt),
+			"encryptedDestination": ctx.encU4(dest),
+		}
+		want[i] = expected
+	}
+
+	factory := NewEngineFactory(ctx.params, ctx.bsk)
+	for _, conc := range []int{1, 4} {
+		conc := conc
+		t.Run(fmt.Sprintf("conc=%d", conc), func(t *testing.T) {
+			res, err := plan.EvaluateBatch(context.Background(), factory, BatchRequest{
+				Inputs:      inputs,
+				Constants:   constants,
+				Concurrency: conc,
+			})
+			if err != nil {
+				t.Fatalf("batch evaluate: %v", err)
+			}
+			if len(res.Verdicts) != N {
+				t.Fatalf("verdicts: got %d want %d", len(res.Verdicts), N)
+			}
+			for i, v := range res.Verdicts {
+				got := ctx.dec.Decrypt(v)
+				if got != want[i] {
+					t.Errorf("conc=%d verdict[%d]: got %v want %v", conc, i, got, want[i])
+				}
+			}
+		})
+	}
+}
+
+// --- Plan cache ------------------------------------------------------
+
+func TestPlanCache_HitsAvoidRecompile(t *testing.T) {
+	yamlBytes := []byte(inlineYAMLU4)
+	hash := HashYAML(yamlBytes)
+
+	var loadCalls int32
+	loader := PlanLoaderFunc(func(id string, h [32]byte) ([]byte, error) {
+		atomic.AddInt32(&loadCalls, 1)
+		return yamlBytes, nil
+	})
+
+	cache := &PlanCache{}
+	_, err := cache.GetOrCompile("treasury-cold-test", hash, loader)
+	if err != nil {
+		t.Fatalf("first GetOrCompile: %v", err)
+	}
+	for i := 0; i < 5; i++ {
+		_, err := cache.GetOrCompile("treasury-cold-test", hash, loader)
+		if err != nil {
+			t.Fatalf("repeated GetOrCompile[%d]: %v", i, err)
+		}
+	}
+	if got := atomic.LoadInt32(&loadCalls); got != 1 {
+		t.Errorf("loader called %d times; want 1", got)
+	}
+	if cache.Len() != 1 {
+		t.Errorf("cache len = %d; want 1", cache.Len())
+	}
+}
+
+func TestPlanCache_RejectsHashMismatch(t *testing.T) {
+	yamlBytes := []byte(inlineYAMLU4)
+	wrongHash := [32]byte{0xDE, 0xAD, 0xBE, 0xEF}
+
+	loader := PlanLoaderFunc(func(id string, h [32]byte) ([]byte, error) {
+		return yamlBytes, nil
+	})
+	cache := &PlanCache{}
+	_, err := cache.GetOrCompile("treasury-cold-test", wrongHash, loader)
+	if err == nil {
+		t.Fatal("expected ErrHashMismatch, got nil")
+	}
+	if !strings.Contains(err.Error(), "policy hash mismatch") {
+		t.Fatalf("err = %v; want hash mismatch", err)
+	}
+}
+
+func TestPlanCache_ForgetEvicts(t *testing.T) {
+	yamlBytes := []byte(inlineYAMLU4)
+	hash := HashYAML(yamlBytes)
+
+	var loadCalls int32
+	loader := PlanLoaderFunc(func(id string, h [32]byte) ([]byte, error) {
+		atomic.AddInt32(&loadCalls, 1)
+		return yamlBytes, nil
+	})
+	cache := &PlanCache{}
+	_, _ = cache.GetOrCompile("treasury-cold-test", hash, loader)
+	cache.Forget("treasury-cold-test")
+	if cache.Len() != 0 {
+		t.Errorf("after forget: len=%d want 0", cache.Len())
+	}
+	_, _ = cache.GetOrCompile("treasury-cold-test", hash, loader)
+	if got := atomic.LoadInt32(&loadCalls); got != 2 {
+		t.Errorf("loader called %d times; want 2 (one before, one after forget)", got)
+	}
+}
+
+func TestPlanCache_DistinctHashesCacheSeparately(t *testing.T) {
+	a := []byte(inlineYAMLU4)
+	b := []byte(strings.Replace(inlineYAMLU4, "treasury-cold-test", "treasury-cold-test", 1) + "\n# trailing change\n")
+	hashA := HashYAML(a)
+	hashB := HashYAML(b)
+	if hashA == hashB {
+		t.Fatal("test setup: a and b must hash differently")
+	}
+	loader := PlanLoaderFunc(func(_ string, h [32]byte) ([]byte, error) {
+		if h == hashA {
+			return a, nil
+		}
+		return b, nil
+	})
+	cache := &PlanCache{}
+	if _, err := cache.GetOrCompile("treasury-cold-test", hashA, loader); err != nil {
+		t.Fatalf("a: %v", err)
+	}
+	if _, err := cache.GetOrCompile("treasury-cold-test", hashB, loader); err != nil {
+		t.Fatalf("b: %v", err)
+	}
+	if cache.Len() != 2 {
+		t.Errorf("len=%d want 2", cache.Len())
+	}
+}

policy/rule_executor.go

@@ -0,0 +1,183 @@
+// Copyright (c) 2026, Lux Industries Inc
+// SPDX-License-Identifier: BSD-3-Clause
+
+package policy
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/luxfi/fhe"
+)
+
+// ErrMissingOperand is returned when an EvaluationStep references a field
+// or constant key that is not present in the input/constant map. Callers
+// must populate every key referenced by the plan; partial maps are not
+// accepted.
+var ErrMissingOperand = errors.New("rule_executor: missing operand")
+
+// Operands bundles the per-call inputs and the per-policy constants used
+// during evaluation. Both maps are read-only during a single Evaluate
+// call; the executor never writes to either.
+//
+// Inputs (per intent): encryptedAmount, encryptedDestination, encryptedSum.
+// Constants (per policy version): encryptedLimit, encryptedAddrN, encryptedVelocityCap.
+//
+// The keys used here MUST match the YAML clause names verbatim. Mismatch
+// returns ErrMissingOperand.
+type Operands struct {
+	Inputs    map[string]*fhe.BitCiphertext
+	Constants map[string]*fhe.BitCiphertext
+}
+
+// Engine bundles the FHE evaluators a RulePlan needs.
+//
+// IMPORTANT: an Engine is NOT goroutine-safe. luxfi/fhe.Evaluator and
+// BitwiseEvaluator hold internal scratch buffers used during blind
+// rotation; concurrent calls into the same Engine race on those
+// buffers. For parallel batch evaluation use EngineFactory and one
+// Engine per worker — see rule_batch.go.
+type Engine struct {
+	Bit   *fhe.BitwiseEvaluator
+	Gates *fhe.Evaluator
+}
+
+// NewEngine wires a fresh BitwiseEvaluator + Evaluator pair from the
+// FHE network public material. The evaluators do not need a secret key.
+func NewEngine(params fhe.Parameters, bsk *fhe.BootstrapKey) *Engine {
+	return &Engine{
+		Bit:   fhe.NewBitwiseEvaluator(params, bsk, nil),
+		Gates: fhe.NewEvaluator(params, bsk),
+	}
+}
+
+// Evaluate runs every step in order, then reduces the per-step verdicts
+// via the plan's Combinator. Returns a single encrypted boolean.
+//
+// A single RulePlan may be Evaluate'd from N goroutines concurrently
+// (the plan is read-only after Compile) — but each goroutine MUST pass
+// its own Engine. See rule_batch.go for the canonical batch executor.
+func (p *RulePlan) Evaluate(ctx context.Context, eng *Engine, ops Operands) (*fhe.Ciphertext, error) {
+	if eng == nil || eng.Bit == nil || eng.Gates == nil {
+		return nil, errors.New("rule_executor: nil engine")
+	}
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+
+	verdicts := make([]*fhe.Ciphertext, len(p.Steps))
+	for i := range p.Steps {
+		v, err := runStep(eng, &p.Steps[i], ops)
+		if err != nil {
+			return nil, fmt.Errorf("rule_executor: step %d (%s): %w", i, p.Steps[i].Name, err)
+		}
+		verdicts[i] = v
+	}
+
+	return reduce(eng, verdicts, p.Combinator)
+}
+
+// runStep dispatches a single EvaluationStep to its FHE primitive and
+// returns the encrypted boolean verdict.
+func runStep(eng *Engine, s *EvaluationStep, ops Operands) (*fhe.Ciphertext, error) {
+	field, ok := ops.Inputs[s.FieldKey]
+	if !ok || field == nil {
+		return nil, fmt.Errorf("%w: input %q", ErrMissingOperand, s.FieldKey)
+	}
+
+	switch s.Op {
+	case OpLt:
+		c, err := lookupConst(ops, s.ConstKey)
+		if err != nil {
+			return nil, err
+		}
+		return eng.Bit.Lt(field, c)
+
+	case OpLte:
+		c, err := lookupConst(ops, s.ConstKey)
+		if err != nil {
+			return nil, err
+		}
+		return eng.Bit.Le(field, c)
+
+	case OpEq:
+		c, err := lookupConst(ops, s.ConstKey)
+		if err != nil {
+			return nil, err
+		}
+		return eng.Bit.Eq(field, c)
+
+	case OpIn:
+		return setMembership(eng, field, ops, s.ConstKeys)
+
+	default:
+		return nil, fmt.Errorf("rule_executor: unknown op %q", s.Op)
+	}
+}
+
+// setMembership computes (field == c1) OR (field == c2) OR ... OR (field == cN).
+// Cost is O(N) bootstraps; callers keep N small (typically <= 32).
+func setMembership(eng *Engine, field *fhe.BitCiphertext, ops Operands, keys []string) (*fhe.Ciphertext, error) {
+	if len(keys) == 0 {
+		return nil, fmt.Errorf("%w: empty constants list", ErrMissingOperand)
+	}
+	var acc *fhe.Ciphertext
+	for i, k := range keys {
+		c, err := lookupConst(ops, k)
+		if err != nil {
+			return nil, err
+		}
+		eq, err := eng.Bit.Eq(field, c)
+		if err != nil {
+			return nil, fmt.Errorf("in[%d] eq: %w", i, err)
+		}
+		if acc == nil {
+			acc = eq
+			continue
+		}
+		acc, err = eng.Gates.OR(acc, eq)
+		if err != nil {
+			return nil, fmt.Errorf("in[%d] or: %w", i, err)
+		}
+	}
+	return acc, nil
+}
+
+// reduce folds N per-step verdicts into one via Combinator. AND for
+// "all must pass", OR for "any may pass". A single-step plan returns
+// that step's verdict unchanged.
+func reduce(eng *Engine, verdicts []*fhe.Ciphertext, comb Combinator) (*fhe.Ciphertext, error) {
+	if len(verdicts) == 0 {
+		return nil, errors.New("rule_executor: empty verdict slice")
+	}
+	if len(verdicts) == 1 {
+		return verdicts[0], nil
+	}
+	acc := verdicts[0]
+	for i := 1; i < len(verdicts); i++ {
+		var err error
+		switch comb {
+		case CombAnd:
+			acc, err = eng.Gates.AND(acc, verdicts[i])
+		case CombOr:
+			acc, err = eng.Gates.OR(acc, verdicts[i])
+		default:
+			return nil, fmt.Errorf("rule_executor: unknown combinator %q", comb)
+		}
+		if err != nil {
+			return nil, fmt.Errorf("reduce[%d]: %w", i, err)
+		}
+	}
+	return acc, nil
+}
+
+// lookupConst returns the ciphertext registered under key, or
+// ErrMissingOperand if absent.
+func lookupConst(ops Operands, key string) (*fhe.BitCiphertext, error) {
+	c, ok := ops.Constants[key]
+	if !ok || c == nil {
+		return nil, fmt.Errorf("%w: constant %q", ErrMissingOperand, key)
+	}
+	return c, nil
+}