TealEngineV13

TealEngine v1.3 extends the parallel evaluation pipeline with pre-evaluation stages (FREEZE rules, NHI verification), post-evaluation hooks (evidence emission, response transformation), automation levels, and PLAN_ONLY mode.

TealEngine v1.3 is backward-compatible with v1.2 policies. Existing evaluate_v12() calls continue to work. The new evaluate() method adds pre/post stages and automation level support.

Installation

pip install tealtiger

Quick Start

import asyncio
from uuid import uuid4
from tealtiger import TealEngine, TealClassifier, TealDrift

engine = TealEngine(
    policy={
        "rules": [
            {
                "condition": {"action": "file.write", "path": {"pattern": "/etc/**"}},
                "automation": "approval_required",
                "approval": {"approvers": ["[email protected]"], "timeout": "10m"},
            },
            {"condition": {"action": "file.read"}, "automation": "auto_allow"},
        ]
    },
    freeze=[
        {
            "id": "FREEZE-001",
            "condition": {"action": "system.shutdown"},
            "reason": "FREEZE_SYSTEM_SHUTDOWN",
            "severity": "critical",
        }
    ],
    modules=[TealClassifier(), TealDrift()],
    mode="ENFORCE",
)

async def main():
    decision = await engine.evaluate(
        request={
            "action": "file.write",
            "tool_params": {"path": "/etc/nginx/nginx.conf", "content": "..."},
            "environment": "production",
        },
        ctx={
            "correlation_id": str(uuid4()),
            "agent_id": "deploy-agent-001",
            "attestation": {"token": "eyJ...", "method": "signed-jwt"},
        },
    )

    if decision.pending:
        print(f"Awaiting approval: {decision.approval.token}")
    else:
        print(f"Decision: {decision.action}")

asyncio.run(main())

Class Signature

class TealEngine:
    def __init__(
        self,
        policy: dict,
        freeze: list[dict] | None = None,
        modules: list[TealModule] | None = None,
        mode: str = "ENFORCE",
        failure_policy: dict | None = None,
        nhi: dict | None = None,
        post_hooks: list[PostEvaluationHook] | None = None,
        approval: dict | None = None,
    ) -> None: ...

    async def evaluate(
        self,
        request: dict,
        ctx: dict,
    ) -> Decision: ...

    async def approve_decision(
        self,
        token: str,
        approved_by: str,
        justification: str,
        conditions: dict | None = None,
    ) -> Decision: ...

    async def reject_decision(
        self,
        token: str,
        rejected_by: str,
        reason: str,
    ) -> Decision: ...

    async def load_policy(self, bundle: dict) -> None: ...

    def get_status(self) -> EngineStatus: ...

    async def evaluate_v12(
        self,
        request: dict,
        ctx: dict,
    ) -> Decision: ...

evaluate()

The primary v1.3 evaluation method. Executes the full pipeline:

FREEZE Check → NHI Verification → Pre-Evaluation → Module Dispatch → Merge → Automation → Post-Hooks → Decision

from uuid import uuid4

decision = await engine.evaluate(
    request={
        "action": "database.migrate",
        "environment": "production",
        "model": "gpt-4o",
    },
    ctx={
        "correlation_id": str(uuid4()),
        "agent_id": "deploy-agent-001",
        "mode": "ENFORCE",
    },
)

PLAN_ONLY Mode

Returns what the governance decision would be without executing enforcement or emitting evidence.

plan = await engine.evaluate(
    request={"action": "database.migrate", "environment": "production"},
    ctx={"correlation_id": str(uuid4()), "mode": "PLAN_ONLY"},
)

# plan.action == "PENDING" (what would happen)
# plan.automation_level == "approval_required"
# plan.executed == False
# plan.plan == True
# No evidence emitted, no approval token created

approve_decision()

Resolves a PENDING decision by providing human approval.

final_decision = await engine.approve_decision(
    token="approval-token-xyz789",
    approved_by="[email protected]",
    justification="Reviewed nginx config change — safe to proceed",
)

# final_decision.action == "ALLOW"
# final_decision.approval.status == "APPROVED"

reject_decision()

Resolves a PENDING decision by rejecting it.

rejection = await engine.reject_decision(
    token="approval-token-xyz789",
    rejected_by="[email protected]",
    reason="Config change not reviewed by security team",
)

# rejection.action == "DENY"
# rejection.approval.status == "REJECTED"

load_policy()

Hot-swap the policy bundle at runtime. FREEZE rules are preserved across swaps.

await engine.load_policy({
    "version": "2.1.0",
    "rules": updated_rules,
    "modules": {"classifier": {"mode": "ensemble_union"}},
    "integrity": "sha256:abc123...",
    "signature": "sig:...",
})

If the bundle includes an integrity hash, TealEngine validates it before applying. A mismatched hash triggers BundleIntegrityError and the swap is rejected.

Post-Evaluation Hooks

Post-evaluation hooks run after the decision is computed but before it’s returned:

from tealtiger import TealEngine, PostEvaluationHook

async def siem_export(decision, ctx):
    await siem_client.emit({
        "event": "governance.deny",
        "correlation_id": ctx["correlation_id"],
        "reason_codes": decision.reason_codes,
        "agent_id": ctx.get("agent_id"),
    })

async def metrics_hook(decision, ctx):
    metrics.increment("governance.evaluations", tags={
        "action": decision.action,
        "automation_level": decision.automation_level,
    })

engine = TealEngine(
    policy=my_policy,
    post_hooks=[
        PostEvaluationHook(id="siem-export", trigger="on_deny", handler=siem_export),
        PostEvaluationHook(id="metrics", trigger="always", handler=metrics_hook),
    ],
)

Error Handling

from tealtiger.errors import (
    TealConfigError,
    FreezeViolationError,
    BundleIntegrityError,
    ApprovalExpiredError,
)

try:
    decision = await engine.evaluate(request, ctx)
except FreezeViolationError as e:
    print(f"FREEZE tamper: {e.rule_id}")
except BundleIntegrityError as e:
    print(f"Bundle integrity: {e.expected} != {e.actual}")
except ApprovalExpiredError as e:
    print(f"Token expired: {e.token}")

Performance

Metric	Target	Notes
Pre-evaluation (FREEZE + NHI)	< 2ms	In-memory checks, no I/O
Module dispatch (parallel)	< 15ms p99	asyncio.gather
Post-hooks	< 5ms	Async, non-blocking
Total evaluate()	< 25ms p99	End-to-end with 4 modules
PLAN_ONLY mode	< 10ms p99	Skips evidence emission

Migration from v1.2

# v1.2 — still works in v1.3
decision = await engine.evaluate_v12(
    request={"content": "..."},
    ctx={"correlation_id": "..."},
)

# v1.3 — new evaluate() with full pipeline
decision = await engine.evaluate(
    request={"action": "tool.invoke", "tool": "web_search", "content": "..."},
    ctx={"correlation_id": "...", "agent_id": "agent-001"},
)

FREEZE Rules — Immutable safety controls
Automation Levels — The four automation levels
NHI Governance — Identity verification pipeline
TealProof — Cryptographic evidence for decisions
TealFlow — Declarative governance workflows
Migration Guide v1.3 — Upgrade path from v1.2

Getting Started

Core Concepts

Governance Domains

Policy Authoring

Architecture

Integrations

Deployment

API Reference

Cookbook

Compliance & GRC

Guides

Reference

Versions

Playground

Dashboard

Blog

Installation

Quick Start

Class Signature

evaluate()

PLAN_ONLY Mode

approve_decision()

reject_decision()

load_policy()

Post-Evaluation Hooks

Error Handling

Performance

Migration from v1.2

​Installation

​Quick Start

​Class Signature

​evaluate()

​PLAN_ONLY Mode

​approve_decision()

​reject_decision()

​load_policy()

​Post-Evaluation Hooks

​Error Handling

​Performance

​Migration from v1.2

​Related Documentation

Installation

Quick Start

Class Signature

evaluate()

PLAN_ONLY Mode

approve_decision()

reject_decision()

load_policy()

Post-Evaluation Hooks

Error Handling

Performance

Migration from v1.2

Related Documentation