Rule Script Engine: Writing Advanced Rules in Python/Groovy
The YAML DSL handles most business rules, but falls short when dealing with complex data transformations, external API calls, or custom algorithms. coomia-dip's FunctionRuntime supports five languages -- Python, Groovy, TypeScript, Kotlin, and WASM -- for writing advanced rule scripts, executed in nsjail sandboxes for security. This article explores the script engine architecture, multi-language runtime design, sandbox security model, script lifecycle management, and integration with the Rete rule engine.
CoomiaPublished on August 27, 202515 min read
Share this articleTwitter / X
“Series: S5 Intelligent Decisions · Article 5 | Level: Advanced | Reading Time: 20 min
Rule Script Engine: Writing Advanced Rules in Python/Groovy
#TL;DR
The YAML DSL handles most business rules, but falls short when dealing with complex data transformations, external API calls, or custom algorithms. coomia-dip's FunctionRuntime supports five languages -- Python, Groovy, TypeScript, Kotlin, and WASM -- for writing advanced rule scripts, executed in nsjail sandboxes for security. This article explores the script engine architecture, multi-language runtime design, sandbox security model, script lifecycle management, and integration with the Rete rule engine.
#1. Script Engine Positioning
#1.1 YAML DSL vs. Script Rules
Code
Rule Complexity Spectrum:
Simple Complex
|------|---------|---------|---------|---------|------>
YAML YAML Script Script Custom
single multi Python Groovy ML Model
condition logic transform DSL
+------------+ +------------------+ +----------------+
| Low-Code | | Script Engine | | ML Layer |
| YAML Rules | | Python/Groovy/TS | | Custom Models |
+------------+ +------------------+ +----------------+
Covers: 80% Covers: 15% Covers: 5%
business rules advanced rules predictive rules
#1.2 Use Cases
| Scenario | YAML Capability | Script Needed | Example |
|---|---|---|---|
| Simple threshold | Fully capable | No | inventory < 100 |
| Multi-condition | Fully capable | No | A AND B OR C |
| Date computation | Limited | Yes | business_day + 3 |
| External API | Not supported | Yes | Exchange rate lookup |
| Statistical agg | Not supported | Yes | Moving average |
| Custom scoring | Limited | Yes | Weighted scoring |
| Recursive logic | Not supported | Yes | Hierarchical approval |
#2. Script Engine Architecture
#2.1 Overall Architecture
Code
Script Engine Architecture:
+----------------------------------------------------------+
| ScriptEngine (entry) |
+----------------------------------------------------------+
| |
| +------------------+ +-------------------+ |
| | Script Registry | | Script Compiler | |
| | (registry) | | (precompile/cache) | |
| +------------------+ +-------------------+ |
| |
| +--------------------------------------------------+ |
| | Runtime Dispatcher | |
| | +----------+ +----------+ +----------+ | |
| | | Python | | Groovy | | TypeScript| | |
| | | Runtime | | Runtime | | Runtime | | |
| | +----------+ +----------+ +----------+ | |
| | +----------+ +----------+ | |
| | | Kotlin | | WASM | | |
| | | Runtime | | Runtime | | |
| | +----------+ +----------+ | |
| +--------------------------------------------------+ |
| |
| +--------------------------------------------------+ |
| | nsjail Sandbox Layer | |
| | - FS isolation - Network isolation - Limits | |
| +--------------------------------------------------+ |
+----------------------------------------------------------+
#2.2 Core Interfaces
Python
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any
from enum import Enum
import time
class ScriptLanguage(str, Enum):
PYTHON = "python"
GROOVY = "groovy"
TYPESCRIPT = "typescript"
KOTLIN = "kotlin"
WASM = "wasm"
@dataclass
class ScriptMetadata:
"""Script metadata"""
script_id: str
name: str
language: ScriptLanguage
version: str
author: str
description: str
inputs: dict[str, str] # name -> type
outputs: dict[str, str] # name -> type
timeout_ms: int = 5000
max_memory_mb: int = 256
tags: list[str] = field(default_factory=list)
@dataclass
class ScriptResult:
"""Script execution result"""
success: bool
outputs: dict[str, Any]
logs: list[str]
elapsed_ms: float
memory_used_mb: float
error: str | None = None
class ScriptRuntime(ABC):
"""Script runtime base class"""
@abstractmethod
def compile(self, source: str, metadata: ScriptMetadata) -> str:
"""Pre-compile script, return compiled identifier"""
...
@abstractmethod
def execute(self, compiled_id: str,
inputs: dict[str, Any]) -> ScriptResult:
"""Execute compiled script"""
...
@abstractmethod
def validate(self, source: str) -> list[str]:
"""Validate script syntax, return error list"""
...
#3. Python Script Runtime
#3.1 Script Writing Standards
Python
# Rule script: Dynamic credit scoring
# File: scripts/credit_scoring.py
from rule_sdk import RuleContext, RuleResult
def evaluate(ctx: RuleContext) -> RuleResult:
"""
Dynamic credit scoring rule.
Computes a composite credit score using
a weighted multi-dimensional model.
"""
credit_score = ctx.get("credit_score", type=int)
income = ctx.get("annual_income", type=float)
debt_ratio = ctx.get("debt_ratio", type=float)
employment_years = ctx.get("employment_years", type=int)
industry = ctx.get("industry", type=str)
# Industry risk coefficients
industry_risk = {
"technology": 0.8,
"finance": 0.9,
"manufacturing": 1.0,
"real_estate": 1.2,
"crypto": 1.5,
}
risk_factor = industry_risk.get(industry, 1.0)
# Weighted scoring
score = (
credit_score * 0.35 +
min(income / 10000, 100) * 0.25 +
(1 - debt_ratio) * 100 * 0.20 +
min(employment_years * 5, 50) * 0.10 +
(1 / risk_factor) * 100 * 0.10
)
if score >= 80:
decision = "approve"
confidence = min(score / 100, 0.99)
elif score >= 60:
decision = "conditional_approve"
confidence = score / 100
else:
decision = "reject"
confidence = (100 - score) / 100
return RuleResult(
decision=decision,
confidence=confidence,
outputs={
"composite_score": round(score, 2),
"risk_factor": risk_factor,
},
explanation=f"Composite score {score:.1f}, industry risk {risk_factor}",
)
#3.2 RuleContext SDK
Python
from dataclasses import dataclass, field
from typing import Any, TypeVar
T = TypeVar("T")
@dataclass
class RuleContext:
"""Execution context for script rules"""
_inputs: dict[str, Any]
_variables: dict[str, Any] = field(default_factory=dict)
_logs: list[str] = field(default_factory=list)
def get(self, name: str, type: type[T] = str,
default: T | None = None) -> T:
"""Get input variable with type checking"""
value = self._inputs.get(name)
if value is None:
if default is not None:
return default
raise KeyError(f"Required input '{name}' not provided")
if not isinstance(value, type):
try:
value = type(value)
except (ValueError, TypeError) as e:
raise TypeError(
f"Variable '{name}' type conversion failed: "
f"expected {type.__name__}, got {value!r}"
) from e
return value
def set_var(self, name: str, value: Any) -> None:
self._variables[name] = value
def get_var(self, name: str, default: Any = None) -> Any:
return self._variables.get(name, default)
def log(self, message: str) -> None:
self._logs.append(message)
@property
def logs(self) -> list[str]:
return list(self._logs)
@dataclass
class RuleResult:
"""Return value from script rules"""
decision: str
confidence: float = 1.0
outputs: dict[str, Any] = field(default_factory=dict)
explanation: str = ""
actions: list[dict] = field(default_factory=list)
#3.3 Python Runtime Implementation
Python
import importlib.util
import tempfile
import subprocess
import json
from pathlib import Path
class PythonScriptRuntime(ScriptRuntime):
"""Python script runtime"""
def __init__(self, sandbox_enabled: bool = True):
self._sandbox = sandbox_enabled
self._compiled: dict[str, Path] = {}
self._cache_dir = Path(tempfile.mkdtemp(prefix="onto_scripts_"))
def compile(self, source: str, metadata: ScriptMetadata) -> str:
errors = self.validate(source)
if errors:
raise ValueError(f"Syntax errors: {errors}")
security_issues = self._security_scan(source)
if security_issues:
raise SecurityError(f"Security violations: {security_issues}")
script_path = self._cache_dir / f"{metadata.script_id}.py"
script_path.write_text(source)
compiled_id = f"{metadata.script_id}:{metadata.version}"
self._compiled[compiled_id] = script_path
return compiled_id
def execute(self, compiled_id: str,
inputs: dict[str, Any]) -> ScriptResult:
script_path = self._compiled.get(compiled_id)
if not script_path:
raise KeyError(f"Script not compiled: {compiled_id}")
start = time.monotonic()
if self._sandbox:
result = self._execute_sandboxed(script_path, inputs)
else:
result = self._execute_direct(script_path, inputs)
result.elapsed_ms = (time.monotonic() - start) * 1000
return result
def _execute_sandboxed(self, script_path: Path,
inputs: dict) -> ScriptResult:
"""Execute via nsjail sandbox"""
input_file = self._cache_dir / "input.json"
input_file.write_text(json.dumps(inputs))
cmd = [
"nsjail",
"--mode", "once",
"--chroot", "/",
"--rlimit_as", "256",
"--rlimit_cpu", "5",
"--rlimit_fsize", "0",
"--disable_clone_newnet",
"--bindmount_ro", f"{script_path}:/script.py",
"--bindmount_ro", f"{input_file}:/input.json",
"--",
"python3", "/script.py", "/input.json",
]
try:
proc = subprocess.run(
cmd, capture_output=True, text=True, timeout=10,
)
if proc.returncode != 0:
return ScriptResult(
success=False, outputs={}, logs=[],
elapsed_ms=0, memory_used_mb=0,
error=proc.stderr,
)
output = json.loads(proc.stdout)
return ScriptResult(
success=True,
outputs=output.get("outputs", {}),
logs=output.get("logs", []),
elapsed_ms=0,
memory_used_mb=output.get("memory_mb", 0),
)
except subprocess.TimeoutExpired:
return ScriptResult(
success=False, outputs={}, logs=[],
elapsed_ms=10000, memory_used_mb=0,
error="Execution timeout (10s)",
)
def _execute_direct(self, script_path: Path,
inputs: dict) -> ScriptResult:
"""Direct execution (dev/test only)"""
spec = importlib.util.spec_from_file_location("rule_script", script_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
ctx = RuleContext(_inputs=inputs)
result = module.evaluate(ctx)
return ScriptResult(
success=True,
outputs={
"decision": result.decision,
"confidence": result.confidence,
**result.outputs,
},
logs=ctx.logs,
elapsed_ms=0,
memory_used_mb=0,
)
def validate(self, source: str) -> list[str]:
try:
compile(source, "<rule_script>", "exec")
return []
except SyntaxError as e:
return [f"Line {e.lineno}: {e.msg}"]
def _security_scan(self, source: str) -> list[str]:
issues = []
forbidden = [
("import os", "os module import forbidden"),
("import sys", "sys module import forbidden"),
("import subprocess", "subprocess module import forbidden"),
("__import__", "Dynamic import forbidden"),
("eval(", "eval() forbidden"),
("exec(", "exec() forbidden"),
("open(", "Direct file operations forbidden"),
]
for pattern, message in forbidden:
if pattern in source:
issues.append(message)
return issues
#4. Groovy Script Runtime
#4.1 Groovy Rule Example
GROOVY
// Rule script: Supply chain risk assessment
// File: scripts/supply_chain_risk.groovy
import onto.rules.RuleContext
import onto.rules.RuleResult
RuleResult evaluate(RuleContext ctx) {
def supplier = ctx.get("supplier_name", String)
def leadTime = ctx.get("lead_time_days", Integer)
def qualityScore = ctx.get("quality_score", Double)
def onTimeRate = ctx.get("on_time_delivery_rate", Double)
def alternativeCount = ctx.get("alternative_suppliers", Integer)
def riskScore = 0.0
// Lead time risk
if (leadTime > 30) riskScore += 25
else if (leadTime > 14) riskScore += 10
// Quality risk
if (qualityScore < 0.8) riskScore += 30
else if (qualityScore < 0.9) riskScore += 15
// On-time delivery risk
if (onTimeRate < 0.85) riskScore += 25
else if (onTimeRate < 0.95) riskScore += 10
// Supplier substitutability risk
if (alternativeCount == 0) riskScore += 20
else if (alternativeCount == 1) riskScore += 10
def decision
def confidence
if (riskScore >= 60) {
decision = "high_risk"
confidence = Math.min(riskScore / 100.0, 0.99)
} else if (riskScore >= 30) {
decision = "medium_risk"
confidence = 0.8
} else {
decision = "low_risk"
confidence = 0.9
}
ctx.log("Supplier ${supplier} risk score: ${riskScore}")
return new RuleResult(
decision: decision,
confidence: confidence,
outputs: [
risk_score: riskScore,
risk_breakdown: [
lead_time: leadTime > 14 ? "elevated" : "normal",
quality: qualityScore < 0.9 ? "concern" : "ok",
delivery: onTimeRate < 0.95 ? "concern" : "ok",
alternatives: alternativeCount < 2 ? "limited" : "ok",
]
],
explanation: "Supplier ${supplier} composite risk score ${riskScore}"
)
}
#4.2 Groovy Runtime Implementation
Python
class GroovyScriptRuntime(ScriptRuntime):
"""Groovy script runtime"""
def __init__(self, groovy_home: str = "/opt/groovy"):
self._groovy_home = groovy_home
self._compiled: dict[str, Path] = {}
def compile(self, source: str, metadata: ScriptMetadata) -> str:
script_path = Path(tempfile.mkdtemp()) / f"{metadata.script_id}.groovy"
script_path.write_text(source)
result = subprocess.run(
[f"{self._groovy_home}/bin/groovyc", str(script_path)],
capture_output=True, text=True,
)
if result.returncode != 0:
raise ValueError(f"Groovy compile error: {result.stderr}")
compiled_id = f"{metadata.script_id}:{metadata.version}"
self._compiled[compiled_id] = script_path
return compiled_id
def execute(self, compiled_id: str,
inputs: dict[str, Any]) -> ScriptResult:
script_path = self._compiled.get(compiled_id)
if not script_path:
raise KeyError(f"Script not compiled: {compiled_id}")
start = time.monotonic()
input_json = json.dumps(inputs)
cmd = [
"nsjail", "--mode", "once",
"--rlimit_as", "512",
"--rlimit_cpu", "10",
"--",
f"{self._groovy_home}/bin/groovy",
"-cp", "/opt/onto/rule-sdk.jar",
str(script_path),
input_json,
]
try:
proc = subprocess.run(
cmd, capture_output=True, text=True, timeout=15,
)
elapsed = (time.monotonic() - start) * 1000
if proc.returncode != 0:
return ScriptResult(
success=False, outputs={}, logs=[],
elapsed_ms=elapsed, memory_used_mb=0,
error=proc.stderr,
)
output = json.loads(proc.stdout)
return ScriptResult(
success=True,
outputs=output.get("outputs", {}),
logs=output.get("logs", []),
elapsed_ms=elapsed,
memory_used_mb=output.get("memory_mb", 0),
)
except subprocess.TimeoutExpired:
return ScriptResult(
success=False, outputs={}, logs=[],
elapsed_ms=15000, memory_used_mb=0,
error="Groovy execution timeout (15s)",
)
def validate(self, source: str) -> list[str]:
result = subprocess.run(
[f"{self._groovy_home}/bin/groovyc", "--dry-run", "-"],
input=source, capture_output=True, text=True,
)
return [result.stderr] if result.returncode != 0 else []
#5. Sandbox Security Model
#5.1 Security Isolation Layers
Code
Security Isolation Architecture:
Layer 1: Static Analysis (compile-time)
+------------------------------------------+
| - Forbidden imports (os, sys, subprocess) |
| - Forbidden eval/exec |
| - No file/network operations |
| - No reflection or dynamic class loading |
+------------------------------------------+
|
v
Layer 2: nsjail Sandbox (runtime)
+------------------------------------------+
| - Separate namespaces (PID, NET, MNT) |
| - Read-only filesystem |
| - No network access |
| - CPU time limit (5-10 seconds) |
| - Memory limit (256-512 MB) |
| - No file creation |
+------------------------------------------+
|
v
Layer 3: Result Validation (return-time)
+------------------------------------------+
| - Output size limit (< 1MB) |
| - Output type validation |
| - Execution time audit |
+------------------------------------------+
#5.2 Resource Limit Configuration
Python
@dataclass
class SandboxConfig:
"""Sandbox configuration"""
max_cpu_seconds: int = 5
max_memory_mb: int = 256
max_output_bytes: int = 1_048_576 # 1MB
max_file_size: int = 0 # No file writing
network_enabled: bool = False
allowed_imports: list[str] = field(default_factory=lambda: [
"math", "decimal", "datetime", "json",
"collections", "itertools", "functools",
"dataclasses", "typing", "enum",
"rule_sdk",
])
class SandboxManager:
"""Sandbox manager"""
def __init__(self, config: SandboxConfig):
self._config = config
def build_nsjail_args(self, script_path: str,
input_path: str) -> list[str]:
return [
"nsjail",
"--mode", "once",
"--rlimit_as", str(self._config.max_memory_mb),
"--rlimit_cpu", str(self._config.max_cpu_seconds),
"--rlimit_fsize", str(self._config.max_file_size),
"--disable_clone_newnet",
"--bindmount_ro", f"{script_path}:/script",
"--bindmount_ro", f"{input_path}:/input.json",
"--cgroup_mem_max",
str(self._config.max_memory_mb * 1024 * 1024),
]
#6. Script Lifecycle Management
#6.1 Lifecycle State Machine
Code
Script Lifecycle:
DRAFT --> VALIDATING --> VALIDATED --> COMPILING --> COMPILED
| | | | |
| v | v v
| INVALID | ERROR TESTING
| | |
| | +---------+--------+
| | | |
| | v v
| | TEST_PASSED TEST_FAILED
| | |
| | v
| +--------> ACTIVE <------- ROLLBACK
| |
| v
+------------------------------------ARCHIVED
#6.2 Script Registry
Python
from datetime import datetime
class ScriptStatus(str, Enum):
DRAFT = "draft"
VALIDATED = "validated"
COMPILED = "compiled"
TESTING = "testing"
ACTIVE = "active"
ARCHIVED = "archived"
ERROR = "error"
@dataclass
class ScriptRecord:
metadata: ScriptMetadata
source: str
status: ScriptStatus
compiled_id: str | None = None
created_at: datetime = field(default_factory=datetime.utcnow)
updated_at: datetime = field(default_factory=datetime.utcnow)
test_results: dict | None = None
error_message: str | None = None
class ScriptRegistry:
"""Script registry"""
def __init__(self):
self._scripts: dict[str, ScriptRecord] = {}
self._active_by_domain: dict[str, list[str]] = {}
def register(self, metadata: ScriptMetadata, source: str) -> str:
key = f"{metadata.script_id}:{metadata.version}"
self._scripts[key] = ScriptRecord(
metadata=metadata, source=source,
status=ScriptStatus.DRAFT,
)
return key
def activate(self, key: str) -> None:
record = self._scripts[key]
record.status = ScriptStatus.ACTIVE
domain = record.metadata.tags[0] if record.metadata.tags else "default"
self._active_by_domain.setdefault(domain, []).append(key)
def get_active_scripts(self, domain: str) -> list[ScriptRecord]:
keys = self._active_by_domain.get(domain, [])
return [self._scripts[k] for k in keys if k in self._scripts]
#7. Integration with Rete Rule Engine
#7.1 Scripts as Rule Actions
Python
class ScriptActionExecutor:
"""Execute scripts as Rete rule actions"""
def __init__(self, engine: "ScriptEngine"):
self._engine = engine
def create_action(self, script_id: str, version: str) -> callable:
compiled_id = f"{script_id}:{version}"
def action(token) -> dict:
inputs = {}
for fact in token.facts:
for k, v in fact.attributes:
inputs[k] = v
result = self._engine.execute(compiled_id, inputs)
return result.outputs if result.success else {"error": result.error}
return action
#7.2 Referencing Scripts from YAML
YAML
apiVersion: rules/v1
kind: RuleSet
metadata:
name: advanced-credit-rules
domain: credit
version: "3.0.0"
spec:
inputs:
credit_score: { type: integer }
annual_income: { type: decimal }
industry: { type: string }
rules:
- id: ACR-001
name: "Dynamic credit scoring (script rule)"
priority: 100
when:
all:
- credit_score >= 500
- annual_income >= 50000
then:
script:
id: credit_scoring
version: "1.2.0"
language: python
input_mapping:
credit_score: "{{ credit_score }}"
annual_income: "{{ annual_income }}"
industry: "{{ industry }}"
#7.3 Hybrid Execution Flow
Code
Hybrid Rule Execution Flow:
Input Facts
|
v
+---+---+
| Rete |----> YAML rule matching
| Net | |
+---+---+ v
| Condition met?
| | |
| Yes No -> next rule
| |
| v
| +-----+------+
| | Action type?|
| +-----+------+
| | |
| v v
| Simple Script
| action action
| (direct) (ScriptEngine)
| | |
| v v
| Result <----+
| |
v v
New facts injected into working memory
#8. Multi-Language Performance Comparison
#8.1 Benchmark Results
Code
Script Execution Latency (milliseconds):
Language Cold Start Hot Exec Throughput(/s)
----------- ---------- -------- --------------
Python | 120 | | 8 | | 125 |
Groovy | 850 | | 12 | | 83 |
TypeScript | 200 | | 5 | | 200 |
Kotlin | 900 | | 10 | | 100 |
WASM | 50 | | 2 | | 500 |
#8.2 Language Selection Guide
| Language | Best For | Advantage | Disadvantage |
|---|---|---|---|
| Python | Data computation, ML | Rich ecosystem | GIL limits concurrency |
| Groovy | Java system integration | JVM interop | Slow cold start |
| TypeScript | Frontend rule consistency | Type safety | V8 memory footprint |
| Kotlin | JVM high-perf rules | Coroutines | Slow cold start |
| WASM | High-frequency low-latency | Ultra-fast execution | Complex to write |
#9. gRPC Interface
PROTOBUF
syntax = "proto3";
package onto.scripts.v1;
service ScriptEngineService {
rpc RegisterScript(RegisterRequest) returns (RegisterResponse);
rpc CompileScript(CompileRequest) returns (CompileResponse);
rpc ExecuteScript(ExecuteRequest) returns (ExecuteResponse);
rpc ValidateScript(ValidateRequest) returns (ValidateResponse);
rpc ListScripts(ListRequest) returns (ListResponse);
}
message RegisterRequest {
string script_id = 1;
string name = 2;
string language = 3;
string version = 4;
string source = 5;
map<string, string> inputs = 6;
map<string, string> outputs = 7;
int32 timeout_ms = 8;
}
message ExecuteRequest {
string compiled_id = 1;
map<string, string> inputs = 2;
}
message ExecuteResponse {
bool success = 1;
map<string, string> outputs = 2;
repeated string logs = 3;
double elapsed_ms = 4;
string error = 5;
}
#10. Best Practices
#10.1 Script Writing Checklist
Code
+---------------------------------------------+
| Script Rule Writing Checklist |
+---------------------------------------------+
| [ ] Function signature: evaluate(ctx) |
| [ ] All inputs via ctx.get() |
| [ ] Type annotations used |
| [ ] Returns RuleResult object |
| [ ] No external I/O (file, network, DB) |
| [ ] No global state mutation |
| [ ] Execution time < 5 seconds |
| [ ] Memory usage < 256 MB |
| [ ] Has unit tests |
| [ ] Has docstring |
+---------------------------------------------+
#10.2 Choosing YAML vs. Script
Code
Decision Tree:
Does the rule involve complex computation?
| |
No Yes
| |
v v
Does it need Use script
external APIs?
| |
No Yes
| |
v v
YAML Script
#Key Takeaways
- Script engine complements the YAML DSL, handling the 15% of rules requiring complex logic
- Five languages (Python/Groovy/TS/Kotlin/WASM) cover different tech stacks and performance needs
- Three-layer security (static analysis + nsjail sandbox + result validation) ensures safe execution
- RuleContext SDK provides type-safe input retrieval and structured result returns
- Integration with Rete engine allows YAML rules to invoke scripts as advanced actions
- WASM runtime latency is just 2ms, ideal for high-frequency low-latency scenarios
- Full lifecycle management (register -> compile -> test -> activate -> archive)
#Next Article
Next up: S5-06 Reasoning Explainability: Why Did the System Make This Decision? will show how to make every reasoning and decision result traceable, explainable, and auditable.
tags: #script-engine #python #groovy #nsjail #sandbox #function-runtime #coomia-dip