Batch
Evaluate a file of records against one compiled bundle, in process or through one engine process.
A batch job evaluates each record on the server channel before writing anything: to re-check stored
records after a rule change, to screen an import, or to compare what two versions of a ruleset
decide. Both forms below run against the published bundle
conformance/bundles/acme.payments.transfer.bundle.json,
the compiled form of the payments contract. Its head says which ruleset and version it is, and the
checksum every result carries:
"ruleCascadeBundle": "1.0.0",
"id": "acme.payments.transfer",
"version": "1.0.0",
"checksum": "sha256:c192dd53b5b1d307d52ccbc27fc1674114e8714d53b699b24088a648ae242c7e",The pipeline
The engine is started once and the bundle loaded once. One thread writes requests while the main thread reads the answers, so the engine never waits for the script and memory stays flat.
Read the input and start one engine
examples/batch/batch.py needs only the Python standard library. A
CSV row becomes an entity: dotted column names build nested objects, and actor.id and
actor.roles become the actor.
id,type,amount,currency,memo,beneficiary.name,beneficiary.country,beneficiary.swiftCode,actor.id,actor.roles
t-1,domestic,120.50,USD,rent,Jo Lee,US,,u-1,teller
t-2,domestic,80,USD,,Ana,US,,u-1,teller
t-3,international,500,USD,gift,X,KP,ABCDKPPY,u-1,teller
t-4,international,900,EUR,invoice 7,Luis,ES,BSCHESMM,u-2,teller
t-5,international,900,EUR,invoice 8,Luis,ES,,u-2,teller
t-6,domestic,-5,USD,refund,Kim,US,,u-3,teller
t-7,domestic,30000,USD,house,Sam,US,,u-9,risk-officerLoad the bundle, then stream the records through it
def run(requests, bundle, engine, env=None, channel="server", out=sys.stdout):
"""Stream requests through one engine. Returns the counts {"allow", "deny", "refused"}."""
process = subprocess.Popen(engine, stdin=subprocess.PIPE, stdout=subprocess.PIPE, env=env,
text=True, encoding="utf-8", bufsize=1)
pending = queue.Queue(maxsize=10_000) # what has been sent and not yet answered, in order
failure = []
def call(request):
process.stdin.write(dumps(request) + "\n")
process.stdin.flush()
line = process.stdout.readline()
if not line:
raise SystemExit("the engine closed its output")
return json.loads(line)
loaded = call({"id": "load", "command": "load", "bundle": bundle})
if not loaded.get("ok"):
process.kill()
raise SystemExit(f"the engine refused the bundle: {loaded.get('error')}")
ruleset = loaded["result"]["ruleset"]
def writer():
try:
for number, request in enumerate(requests, start=1):
pending.put((number, request))
process.stdin.write(dumps({"id": number, "command": "evaluate", "ruleset": ruleset,
"channel": channel, "request": request}) + "\n")
process.stdin.close()
except BaseException as e: # a bad input line; reported after the answers already in flight
failure.append(e)
try:
process.stdin.close()
except OSError:
pass
finally:
pending.put(None)
thread = threading.Thread(target=writer, daemon=True)
thread.start()
counts = {"allow": 0, "deny": 0, "refused": 0}
while True:
item = pending.get()
if item is None:
break
number, request = item
line = process.stdout.readline()
if not line:
raise SystemExit("the engine stopped answering")
response = json.loads(line)
if response.get("id") != number:
raise SystemExit(f"the engine answered out of order: expected id {number}, got {response.get('id')}")
decision = summary(number, request, response)
counts[decision.get("decision", "refused")] += 1Run it
With the rcas command as the engine (any engine works: Python, TypeScript, Java or the
WebAssembly module, and the decisions are the same):
python3 examples/batch/batch.py examples/batch/transfers.csv --engine "rcas engine"{"line": 1, "id": "t-1", "decision": "allow", "blocking": [], "findings": [], "commands": []}
{"line": 2, "id": "t-2", "decision": "allow", "blocking": [], "findings": [{"code": "ORG-TRF-003", "severity": "warning", "status": "open", "message": "Adding a memo makes this transfer easier to reconcile."}], "commands": []}
{"line": 3, "id": "t-3", "decision": "deny", "blocking": ["ORG-TRF-001"], "findings": [{"code": "ORG-TRF-001", "severity": "error", "status": "open", "message": "Transfers to KP are not permitted."}], "commands": []}
{"line": 4, "id": "t-4", "decision": "allow", "blocking": [], "findings": [], "commands": []}
{"line": 5, "id": "t-5", "decision": "deny", "blocking": ["PAY-TRF-002"], "findings": [{"code": "PAY-TRF-002", "severity": "error", "status": "open", "message": "A valid SWIFT/BIC code is required for international transfers."}], "commands": []}
{"line": 6, "id": "t-6", "decision": "deny", "blocking": ["PAY-TRF-001"], "findings": [{"code": "PAY-TRF-001", "severity": "error", "status": "open", "message": "Amount must be greater than zero."}], "commands": []}
{"line": 7, "id": "t-7", "decision": "deny", "blocking": ["ORG-TRF-002", "PAY-TRF-003"], "findings": [{"code": "ORG-TRF-002", "severity": "error", "status": "open", "message": "Amount exceeds the single-transfer limit of 25000."}, {"code": "PAY-TRF-003", "severity": "warning", "status": "open", "message": "This is a large transfer to Sam. Please confirm the details."}], "commands": []}
7 request(s): 3 allowed, 4 denied, 0 refused as malformedDone when the summary reads 3 allowed, 4 denied, 0 refused as malformed and the exit status is
0. The exit status is 1 when a request was refused as malformed, and 2 when the engine could not be
started or broke the protocol.
In process, without an engine
When the job is written in a language with a runtime, evaluate in process instead:
rules = RuleSet.from_bundle(json.loads(Path(bundle_path).read_text(encoding="utf-8")))
now = "2026-10-03T00:00:00Z" # one clock for the whole run: results are reproducible
for number, line in enumerate(records, start=1):
result = rules.evaluate({
"entity": "Transfer",
"operation": "create",
"data": json.loads(line),
"actor": {"id": "batch-import", "roles": ["system"]},
"ctx": {"now": now},
})The complete in-process script, its input and its verified output are in the batch processing playbook.
For throughput, the engine clients README has the measurements: about 0.14 ms per evaluation with one command process kept alive (about 7,000 per second), against 3.4 ms when a process is started per request. Run a pool of processes, one request in flight each, for parallel work.