Python

HelixorEngine() → HelixorEngine

Creates an engine with the built-in example pack (data protection, compliance.regulatory_pii_guard.v1) and the bundled Community license. Needs no files, key or network. The examples on this page use that pack; the API is the same for any pack.

The constructor also accepts an optional guard argument. It is reserved for internal use; leave it out.

engine = HelixorEngine()
print(engine.pack_id, engine.version, engine.tier)
# compliance.regulatory_pii_guard.v1 1.0.0 COMMUNITY

@classmethod load_pack(pack_path: str | PathLike | None = None, *, license_file: str | PathLike | None = None) → HelixorEngine

Loads a compiled .hxpack and unseals it in memory with your .hxlic. With no arguments it returns the same engine as HelixorEngine(). Once you give a pack path, the pack is unsealed or the call raises; it never falls back to the built-in example pack.

pack_pathstr | PathLike | None

Path to a compiled pack; ~ is expanded. See Packs and manifest.

license_filestr | PathLike | None

Path to your license file. Keyword-only. When omitted, the license is searched for in HELIXOR_LICENSE_FILE, then ./helixor.hxlic, then ~/.helixor/helixor.hxlic; the older name helixor.lic is accepted in those two directories. See where the runtime looks.

engine = HelixorEngine.load_pack("internal_ids.hxpack", license_file="~/.helixor/helixor.hxlic")
engine.pack_id      # 'custom.internal_ids.v1'
engine.tier         # 'DEVELOPER'

Raises FileNotFoundError when the pack does not exist; LicenseNotFoundError (LICENSE_NOT_FOUND, also a FileNotFoundError) when the explicit license path or HELIXOR_LICENSE_FILE names a missing file or no license is found, listing the locations checked; LicenseLocationAmbiguousError when one directory holds both helixor.hxlic and helixor.lic; ValueError for license_file without pack_path; PackUnsealError when the pack is invalid, tampered or keyed to another license, UnsupportedPackDeclarationError when the pack declares something the engine does not execute, and a license error when the signature, dates or entitlements fail.

evaluate(text: str | dict) → EvaluationResult

Runs the pack's extraction steps, checks its rules, builds the remedy and returns the result. The call is synchronous and does no I/O.

textstr | dictrequired

The payload. For a dict, it must have exactly one key, one of text, message or payload, whose value is a string (it may be empty). A dict with more than one key raises AmbiguousPayloadError (a CodonRuntimeError, with the keys in .keys, sorted). A dict with none of those keys, or whose value is not a string, raises CodonRuntimeError. Changed in 0.3.0: 0.2.1 read the first of the three keys that was present and ignored the other keys.

Rule order. Every rule is checked, in a fixed order, and every match is listed in triggers. The first fatal match sets action and reason; with no fatal match, the first warning does. The remedy covers everything that was found. In the example pack the order is Social Security number, card number, health record ID, then contact details (email, phone, IPv4), and the remedy redacts every category that was found.

result = engine.evaluate("Employee note: SSN is 123-45-6789, email jane@example.com.")
result.action              # 'block_glba_ssn_leakage'
result.remedy.clean_text   # 'Employee note: SSN is [REDACTED_SSN], email [REDACTED_EMAIL].'
[t.rule_id for t in result.triggers]  # ['RULE-GLBA-SSN-BLOCK', 'RULE-GDPR-EMAIL-REDACT']

A compiled pack evaluates your rules first, then the built-in checks, with the same precedence. See Evaluating decisions.

compile_custom_rule(instruction: str) → RemovedCapabilityError

Removed in 0.3.0. No public engine could run it: HelixorEngine() is Community, and an engine from load_pack() is sealed. hasattr(engine, "compile_custom_rule") is False, and a call raises RemovedCapabilityError, an EnclaveCapabilityError that is also an AttributeError, whose message points to the playbook's rules.

Put custom patterns in a playbook's rules, compile it, and load it with load_pack(). See Playbook schema.

create_streaming_session(lookahead_chars: int = 28, block_on_fatal: bool = True) → StreamingPIISession

Returns a stateful session that you feed token fragments to. It holds back a lookahead window so a value split across fragments ("123", "-45", "-6789") is caught before any part of it is released.

lookahead_charsint

How many trailing characters are held back. Values below 16 are raised to 16. Default 28.

block_on_fatalbool

When True, a fatal match halts the session. When False, fatal matches are remedied (in the example pack, redacted) and the stream continues. Default True.

stream_filter(token_stream: Iterable[str], lookahead_chars: int = 28, block_on_fatal: bool = True) → Iterator[str]

Wraps a synchronous iterable of token strings and yields filtered text. It creates a session, pushes each token and flushes at the end.

On a fatal match with block_on_fatal=True it raises StreamBlockedError("Stream blocked by statutory invariant: <reason> (<action>)"), with action and reason attributes. It subclasses RuntimeError. Text already yielded stays yielded; nothing from the blocked window is released.

from helixor_runtime import StreamBlockedError

try:
    for text in engine.stream_filter(model_tokens()):
        send_to_client(text)
except StreamBlockedError as exc:
    send_to_client("[response withheld]")
    log.warning("stream blocked: %s", exc.action)

astream_filter(token_stream: AsyncIterable[str], lookahead_chars: int = 28, block_on_fatal: bool = True) → AsyncIterator[str]

The async version of stream_filter. Use async for. It raises the same StreamBlockedError on a fatal block.

EvaluationResult

The object evaluate() returns, exported as helixor_runtime.EvaluationResult. Use it for type hints and isinstance checks. (helixor_runtime.DecisionResult is a different class, the generated SDK model described below.) It is a Pydantic model, so result.model_dump() gives the same JSON the HTTP API returns.

Field	Type	Meaning
pack_id	str	The pack that decided.
action	str	One of the pack's declared actions. Values for the example pack: permit_clean_payload, block_glba_ssn_leakage, block_pci_dss_pan_leakage, block_hipaa_phi_leakage, redact_and_permit_contact_pii. A compiled pack adds block_<id> and redact_<id> for its own rules.
invariants_passed	bool	True when no rule fired. False for both fatal and warning results.
reason	str	A readable explanation of the deciding rule.
triggers	list[RegulatoryTrigger]	Every rule that fired, in evaluation order.
remedy	CounterfactualRemedy	The repaired payload. Always present.
latency_us	float	Microseconds spent extracting, checking rules and building the remedy, rounded to 2 places. License checks, receipt hashing and building this object are not included.
tokens_spent	int	Always 0.
egress_bytes	int	Always 0.
receipt_hash	str	hx_proof_ followed by 24 hex characters: a truncated SHA-256 over the helixor.decision_receipt.v1 body: schema, pack ID, pack version, action, outcome, fired rule IDs and a SHA-256 of the input. The built-in pack and compiled packs use the same recipe (changed in 0.3.0). It is deterministic, unkeyed and not chained. See Receipts.
license_status	str | None	COMMUNITY_PERPETUAL, ACTIVE, EXPIRING_SOON or IN_GRACE_PERIOD.
active_key_id	str | None	The ID of the license key that authorized this evaluation.
active_epoch	int | None	That key's rotation epoch.
edition	str	COMMUNITY, DEVELOPER, LAUNCH_TRIAL or ENTERPRISE.
upgrade_notice	str | None	Human-readable licensing text. Do not parse it.

RegulatoryTrigger

Field	Type	Meaning
rule_id	str	For example RULE-GLBA-SSN-BLOCK, RULE-PCI-DSS-PAN-BLOCK, RULE-HIPAA-PHI-BLOCK, RULE-GDPR-EMAIL-REDACT, RULE-TCPA-PHONE-REDACT, RULE-GDPR-IP-REDACT.
law	str	The regulatory framework the rule enforces, as display text.
severity	str	FATAL (do not let the payload proceed) or WARNING (proceed with the remedy).
matched_items	list[str]	The exact values that matched; with the example pack, the sensitive values found.

CounterfactualRemedy

Field	Type	Meaning
clean_text	str	The repaired input. For the example pack, the input with each match replaced by its redaction token: [REDACTED_SSN], [REDACTED_CARD_PAN], [REDACTED_HEALTH_ID], [REDACTED_EMAIL], [REDACTED_PHONE], [REDACTED_IP].
redactions_count	int	How many distinct values were replaced.
redacted_categories	list[str]	Display names of the categories found, for example (example pack) SSN (GLBA/FCRA), Email (GDPR/CCPA).

A remedy is computed even for fatal results. It tells you what was found; it does not make a fatal payload safe to send.

StreamingPIISession

Returned by create_streaming_session(). The held-back buffer stays raw, and each push re-checks all of it.

push(delta: str)→ list[StreamChunkResult]

Adds a fragment and returns zero or more chunks that are safe to emit. Text is released up to the last delimiter (whitespace or . , ; ! ? ( ) [ ] { }) that lies outside the lookahead window and does not cut a value in two: the released part and the held part, redacted separately, must give the same text as the whole buffer redacted. If there is no delimiter and the buffer exceeds twice the window, the latest such cut that keeps the window is used. After a block, push returns []. It does not raise.

flush()→ list[StreamChunkResult]

Checks and releases the rest of the buffer, with is_final=True. Call it once when the source stream ends.

is_blockedbool

True once a fatal match has halted the session.

session = engine.create_streaming_session(lookahead_chars=28, block_on_fatal=True)
for token in model_tokens():
    for chunk in session.push(token):
        if chunk.blocked:
            abort(chunk.action, [t.rule_id for t in chunk.triggers])
            break
        emit(chunk.text)
    if session.is_blocked:
        break
else:
    for chunk in session.flush():
        emit(chunk.text)

The concatenated chunks equal evaluate(full_text).remedy.clean_text for any fragment size. In 0.2.0, short fragments could produce text such as [REDACTED_EMAIL]m; 0.2.1 and later fix that.

StreamChunkResult

Field	Type	Meaning
chunk_index	int	Increases by one per emitted chunk, starting at 1.
text	str	Text to emit. Empty when blocked.
is_final	bool	True on chunks from flush().
redacted	bool	True when this chunk's text differs from the raw text it replaces.
blocked	bool	True on the single chunk that halts the stream.
action	str	stream_token for normal chunks; the pack action on a blocked chunk.
reason	str | None	Set on a blocked chunk.
triggers	list[RegulatoryTrigger]	The fatal triggers, on a blocked chunk only. matched_items holds raw values.

start_server(engine: HelixorEngine | None = None, host: str = "127.0.0.1", port: int = 0, log_level: str = "warning", token: str | None = None) → tuple[Server, Thread, str]

Starts the service on a daemon thread, serving engine. It waits up to 5 seconds for GET /v1/health to answer, then returns (server, thread, base_url), for example http://127.0.0.1:53817. If the service does not answer in time, it stops it and raises ServerConfigurationError.

engineHelixorEngine | None

The engine every endpoint evaluates with, for example one from load_pack(). None serves the built-in example pack. Anything that is not a HelixorEngine raises TypeError.

hoststr

Bind address. A non-loopback address without a token raises ServerConfigurationError.

portint

0 picks a free port.

log_levelstr

Server log level.

tokenstr | None

Bearer token every endpoint except GET /v1/health requires. None reads HELIXOR_SERVICE_TOKEN; if that is unset too, the service has no authentication.

The same service runs as a module: python -m helixor_runtime.server [--pack PACK] [--license LICENSE] [--host 127.0.0.1] [--port 18734] [--token TOKEN] [--log-level info]. Without --pack it serves the built-in example pack.

stop_server(server: Server, thread: Thread) → None

Asks the server to exit and waits up to 2 seconds for the thread.

server, thread, base_url = start_server(host="127.0.0.1", port=0)
try:
    ...  # call base_url + "/v1/evaluate"
finally:
    stop_server(server, thread)

RegulatoryPiiGuardClient(pack_path=None, license_json=None, license_file=None, native_lib_path=None, allow_python_fallback=False, execution_mode=None)

Methods and properties:

• decide(state) returns a DecisionResult with pack_id, action, verdict, confidence, proof_sha256, latency_us, execution_mode, necessary_causes, counterfactual and is_approved. There is no action argument: the first hard rule that fires sets action, otherwise it is permit_clean_payload. verdict is approve or refuse, and is_approved agrees with it. Each cause names its rule and rule_verdict, and counterfactual.changes has one change per cause. A state missing a field a rule reads raises DecisionStateError.
• decide_batch(states) calls decide once per state.
• get_audit_report() returns a TrueUpReport from the native ledger. Without one it raises AuditReportUnavailableError.
• is_native reports whether a native library was loaded; audit_available whether get_audit_report() can answer.

The client runs on a native runtime library. No native library ships with 0.3.1 (Planned). execution_mode is native by default, so the constructor raises NativeRuntimeUnavailableError. Pass execution_mode="python" to use the generated Python evaluator, or allow_python_fallback=True, which selects native_or_python. On the Python evaluator is_native and audit_available are False and each result's execution_mode is embedded_python. A pack_path that does not exist raises FileNotFoundError.