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 read from HELIXOR_LICENSE_FILE, then ~/.helixor/helixor.lic.

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

Raises FileNotFoundError when the pack or the license file does not exist, ValueError for license_file without pack_path, PackUnsealError when no license can be found or 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, the engine reads the first of text, message or payload that is present, in that order, even if it is empty. Other keys are ignored. A dict with none of them, or whose value is not a string, raises CodonRuntimeError. In the next release a dict must have exactly one key, one of those three; a dict with more keys raises AmbiguousPayloadError (a CodonRuntimeError, with the keys in .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) → None

Adds a blocking rule to this engine, in memory only. It needs an engine with a Developer, Launch Trial or Enterprise license that is not a sealed pack. On the Community license it raises CommunityEditionRestrictionError; on an engine from load_pack() it raises EnclaveCapabilityError, because a sealed pack's rules are fixed when it is compiled.

instructionstrrequired

A sentence containing the literal to block. The engine takes the first quoted string ('…' or "…"), or the last word if nothing is quoted. For example, "Block any text containing 'PROJ-ZEUS'" blocks PROJ-ZEUS.

The rule is a case-sensitive substring match, not a regular expression or natural-language interpretation. It gets the rule ID RULE-CUSTOM-<LITERAL> and the action block_custom_rule_<literal>, with severity FATAL. Punctuation in the literal becomes _. Custom rules are checked before the built-in rules, so a custom match decides even when the payload also contains contact details or a health record ID.

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. Custom rules add block_custom_rule_….
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 pack ID, action, outcome, fired rule IDs and a SHA-256 of the input. For a compiled pack it covers the pack ID, action, outcome and remedy.clean_text instead, so inputs with the same action and clean text share a receipt (a known issue in 0.2.1, fixed in the next release, where both engines use helixor.decision_receipt.v1, which adds the pack version). 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 fixes 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)

Methods:

• decide(state, action="permit_clean_payload") returns a DecisionResult with verdict, confidence, proof_sha256, latency_us, execution_mode, necessary_causes, counterfactual and is_approved.
• decide_batch(states, action=…) calls decide once per state.
• get_audit_report() returns a TrueUpReport.
• is_native reports whether a native library was loaded.

The client runs on a native runtime library. No native library ships with 0.2.1 (Planned), so by default the constructor raises NativeRuntimeUnavailableError. Pass allow_python_fallback=True to use the generated Python evaluator instead; is_native is then False and execution_mode is embedded_python. A pack_path that does not exist raises FileNotFoundError.