Evaluating decisions
evaluate() is the one call you make on every payload. This page covers what goes in, what comes out, how to act on it, and what to keep out of your logs.
About the example
This page uses the built-in data-protection pack so you can run everything without writing a pack first. The same calls work for any decision pack: eligibility, limits, routing and so on. See Core concepts.
Create one engine and reuse it#
Construct the engine once, at startup, and keep it for the life of the process. Construction verifies the license; evaluation then only runs the pack.
from helixor_runtime import HelixorEngine ENGINE = HelixorEngine() # once per process
One engine per worker process
The engine keeps usage counters (and, on rate-limited tiers, a request window) that are not synchronized between threads. Give each worker process its own engine and avoid sharing one engine across a thread pool. See Reliability.
Input#
evaluate() accepts a string, or a dictionary with the text under one of these keys (first match wins): text, message, payload.
ENGINE.evaluate("Call me at (415) 555-0182.")
ENGINE.evaluate({"message": "Call me at (415) 555-0182."})
Only one field is evaluated
The runtime evaluates the first of text, message and payload that is present, even if it is empty, and nothing else. {"text": "hi", "message": "SSN 123-45-6789"} is permitted. A dictionary with none of those keys, such as {"body": "SSN 123-45-6789"}, or with a value that is not a string, raises CodonRuntimeError instead of being evaluated. Pass strings, and choose the field yourself.
Fixed in the next release: a dictionary must have exactly one key, one of text, message, payload. A dictionary with more keys, such as {"text": "hi", "message": "SSN 123-45-6789"} or {"message": "…", "channel": "chat"}, raises AmbiguousPayloadError, a CodonRuntimeError whose keys attribute lists the keys.
In 0.2.1, other keys in the dictionary are ignored. To check several fields of a record, evaluate each field on its own; that way the result tells you which field needs attention (see Batch processing).
Output#
The result is an EvaluationResult (exported from helixor_runtime) with three groups of fields.
The decision#
| Field | Type | Meaning |
|---|---|---|
action | str | The chosen action, from the pack's declared list. |
invariants_passed | bool | True when no rule fired. |
reason | str | Human-readable explanation from the deciding rule. |
triggers | list | Every rule that fired, in evaluation order: rule_id, law, severity (FATAL or WARNING), matched_items. |
remedy | object | The repaired payload and what was changed: clean_text, redactions_count, redacted_categories. |
Evidence#
| Field | Meaning |
|---|---|
pack_id | The pack that decided. |
receipt_hash | Fingerprint of this decision. See Receipts. |
latency_us | Evaluation time in microseconds: extraction through remedy. It excludes license checks, receipt hashing and building the result object, so wall-clock time per call is higher. |
tokens_spent, egress_bytes | Always 0. |
License context#
| Field | Meaning |
|---|---|
edition | COMMUNITY, DEVELOPER, LAUNCH_TRIAL or ENTERPRISE. |
license_status | COMMUNITY_PERPETUAL, ACTIVE, EXPIRING_SOON or IN_GRACE_PERIOD. Alert on the last two. |
active_key_id, active_epoch | Which license key decided. |
upgrade_notice | Tier guidance text. Do not show it to end users. |
Precedence: one decision, a complete remedy#
A pack checks every rule, in a fixed order, and reports every match in triggers. The first fatal match decides the action; if nothing fatal matched, the first warning decides. A later, weaker match never turns a block into a redaction. The remedy covers everything the pack found.
In the example pack, the order is Social Security number, card number, health record ID, then email, phone and IP address, and the remedy redacts every category it found:
r = ENGINE.evaluate("Card 4111-1111-1111-1111, reply to ops@example.com")
r.action # 'block_pci_dss_pan_leakage'
[t.rule_id for t in r.triggers] # ['RULE-PCI-DSS-PAN-BLOCK', 'RULE-GDPR-EMAIL-REDACT']
r.remedy.redacted_categories # ['Credit Card PAN (PCI-DSS)', 'Email (GDPR/CCPA)']
r.remedy.clean_text # 'Card [REDACTED_CARD_PAN], reply to [REDACTED_EMAIL]'
Use action and reason to explain the decision, and triggers or remedy.redacted_categories to report everything that was found.
Acting on the result#
Because a pack declares a closed set of actions, you can dispatch on them exhaustively. In the example pack, every action falls into one of three classes: permit, redact and block. Handle all three explicitly and treat anything else as a block, so a new action in a future pack version fails closed. Check trigger severity first: a FATAL trigger always means stop, whatever the action is called.
def route(text: str) -> str | None:
"""Return text that is safe to send onward, or None to stop."""
r = ENGINE.evaluate(text)
fatal = any(t.severity == "FATAL" for t in r.triggers)
if not fatal and r.action.startswith("permit_"):
return text
if not fatal and r.action.startswith("redact_"):
return r.remedy.clean_text
# fatal, block_*, or anything unrecognized: do not forward, not even redacted
audit_log.warning("blocked", extra={"receipt": r.receipt_hash, "action": r.action,
"rules": [t.rule_id for t in r.triggers]})
return None
What not to log#
matched_items contains the matched values
triggers[].matched_items holds the exact substrings that matched; in the example pack, the card number or the SSN. Never log, store or return the whole result object. Log receipt_hash, action, rule IDs and counts instead.
def audit_fields(r):
return {
"receipt": r.receipt_hash,
"pack": r.pack_id,
"action": r.action,
"rules": [t.rule_id for t in r.triggers],
"redactions": r.remedy.redactions_count,
"latency_us": r.latency_us,
}
Errors#
For the Community tier, evaluate() does not raise on string content. It raises CodonRuntimeError for a dictionary it cannot read (see Input). On rate-limited tiers it can raise DeveloperQuotaExceededError when the license's per-minute or lifetime limit is reached and the license enforces limits strictly. Every runtime error derives from HelixorRuntimeError, exported from helixor_runtime. Catch it and fail closed:
from helixor_runtime import HelixorRuntimeError
try:
r = ENGINE.evaluate(text)
except HelixorRuntimeError as exc: # any failure to decide is a block
audit_log.error("decision_failed", extra={"error": type(exc).__name__})
return None
The full list is in Errors.