EnforceCore — API Design (v1.0 – v1.11)

from enforcecore import (
    # Primary API
    enforce,                    # Decorator (sync + async)
    Enforcer,                  # Main class for programmatic control

    # Policy
    Policy,                    # Policy model
    PolicyEngine,              # Rule evaluation engine
    load_policy,               # Load policy from YAML file

    # Config
    Settings,                  # Global configuration
    settings,                  # Module-level singleton

    # Redactor (v1.0.1)
    Redactor,                  # PII detection and redaction engine
    RedactionResult,           # Result of a redaction operation
    DetectedEntity,            # A PII entity found in text

    # Auditor (v1.0.2)
    Auditor,                   # Merkle-chained audit trail writer
    AuditEntry,                # Single audit entry with SHA-256 hash
    VerificationResult,        # Result of trail verification
    verify_trail,              # Verify integrity of a JSONL trail file
    load_trail,                # Load trail entries from a JSONL file

    # Guard (v1.0.3)
    ResourceGuard,             # Time/memory limit enforcer
    CostTracker,               # Thread-safe cumulative cost tracker
    KillSwitch,                # Hard termination on limit breach

    # Integrations (v1.0.4)
    require_package,           # Verify optional deps with install messages
    wrap_with_policy,          # Wrap any callable with enforcement

    # Hardening (v1.0.6)
    validate_tool_name,        # Validate tool name charset/length
    check_input_size,          # Reject oversized payloads
    deep_redact,               # Recursive nested PII redaction
    enter_enforcement,         # Enter enforcement scope (ContextVar)
    exit_enforcement,          # Exit enforcement scope
    get_enforcement_depth,     # Current nesting depth
    get_enforcement_chain,     # List of tool names in scope
    is_dev_mode,               # Check ENFORCECORE_DEV_MODE
    warn_fail_open,            # Emit RuntimeWarning for fail_open

    # Unicode hardening (v1.0.6)
    normalize_unicode,         # NFC normalization + strip zero-width
    normalize_homoglyphs,      # Confusable char replacement
    decode_encoded_pii,        # URL/HTML entity decoding
    prepare_for_detection,     # Full unicode pipeline

    # Exceptions
    EnforceCoreError,          # Base exception
    EnforcementViolation,      # Policy violation (call blocked)
    ToolDeniedError,           # Specific: tool not allowed
    CostLimitError,            # Specific: cost budget exceeded
    ResourceLimitError,        # Specific: resource limit breached
    RedactionError,            # PII redaction error (fails closed)
    HardeningError,            # Base hardening error (v1.0.6)
    InvalidToolNameError,      # Invalid tool name (v1.0.6)
    InputTooLargeError,        # Payload exceeds limit (v1.0.6)
    EnforcementDepthError,     # Recursive depth exceeded (v1.0.6)

    # Plugin hooks (v1.0.7)
    HookContext,               # Pre/post call hook context
    HookRegistry,              # Hook registration and firing
    ViolationHookContext,      # Violation hook context
    RedactionHookContext,      # Redaction hook context
    on_pre_call,               # Decorator: register pre-call hook
    on_post_call,              # Decorator: register post-call hook
    on_violation,              # Decorator: register violation hook
    on_redaction,              # Decorator: register redaction hook

    # Custom patterns (v1.0.7)
    CustomPattern,             # User-defined PII pattern
    PatternRegistry,           # Pattern registration and lookup

    # Secret detection (v1.0.7)
    DetectedSecret,            # A secret found in text
    SecretScanner,             # Scans text for API keys, tokens, etc.

    # Audit backends (v1.0.7)
    AuditBackend,              # ABC for custom audit backends
    JsonlBackend,              # Default JSONL file backend
    NullBackend,               # Discard backend (testing)
    CallbackBackend,           # Send entries to a callable
    MultiBackend,              # Fan-out to multiple backends

    # Types
    CallContext,               # Immutable per-call context
    Decision,                  # Enum: allowed/blocked/redacted
    EnforcementResult,         # Result metadata from an enforced call
    RedactionEvent,            # Record of a single redacted entity
    RedactionStrategy,         # Enum: mask/hash/remove/placeholder
    ViolationType,             # Enum: why a call was blocked
    ViolationAction,           # Enum: block/log/redact
)

from enforcecore import enforce

# Minimal usage — policy from file
@enforce(policy="policies/agent.yaml")
async def search_web(query: str) -> str:
    return await api.search(query)

# Sync functions also work
@enforce(policy="policies/agent.yaml")
def read_file(path: str) -> str:
    return open(path).read()

# Inline policy (for quick prototyping)
@enforce(
    allowed_tools=["search_web", "calculator"],
    pii_redaction=True,
    max_cost_usd=5.0,
)
async def my_tool(args: dict) -> str:
    ...

# With explicit tool name (when function name differs)
@enforce(policy="policy.yaml", tool_name="web_search")
async def _internal_search(query: str) -> str:
    ...

from enforcecore import enforce

async def run_agent_tool(tool_name: str, args: dict) -> dict:
    async with enforce(policy="policy.yaml", tool_name=tool_name) as ctx:
        # ctx.check_pre(args) is called automatically on entry
        result = await execute_tool(tool_name, args)
        ctx.record_result(result)
        # ctx.check_post(result) is called automatically on exit
    return result

from enforcecore import Enforcer, Policy

# Create from a policy
enforcer = Enforcer(Policy.from_file("policy.yaml"))

# Or from a file directly
enforcer = Enforcer.from_file("policy.yaml")

# Sync enforcement
result = enforcer.enforce_sync(search_fn, "query", tool_name="search_web")

# Async enforcement
result = await enforcer.enforce_async(search_fn, "query", tool_name="search_web")

# Properties
print(enforcer.policy_name)  # "my-policy"
print(enforcer.policy)       # Policy instance

from enforcecore import Policy, load_policy

# From YAML file
policy = load_policy("policies/strict.yaml")

# From dict (programmatic)
policy = Policy(
    name="my-policy",
    rules={
        "allowed_tools": ["search", "calculator"],
        "pii_redaction": {"enabled": True, "categories": ["email", "phone"]},
        "resource_limits": {"max_call_duration_seconds": 30},
    },
    on_violation="block",
)

# Validate without loading
errors = Policy.validate_file("policies/strict.yaml")
if errors:
    print(f"Policy has errors: {errors}")

from pydantic import BaseModel
from typing import Literal

class PIIRedactionConfig(BaseModel):
    enabled: bool = False
    categories: list[str] = ["email", "phone", "ssn", "credit_card"]
    strategy: Literal["mask", "hash", "remove", "placeholder"] = "placeholder"

class ResourceLimits(BaseModel):
    max_call_duration_seconds: float | None = None
    max_memory_mb: int | None = None
    max_cost_usd: float | None = None

class NetworkPolicy(BaseModel):
    allowed_domains: list[str] = []
    deny_all_other: bool = True

class PolicyRules(BaseModel):
    allowed_tools: list[str] | None = None    # None = all allowed
    denied_tools: list[str] = []
    pii_redaction: PIIRedactionConfig = PIIRedactionConfig()
    resource_limits: ResourceLimits = ResourceLimits()
    network: NetworkPolicy = NetworkPolicy()
    max_output_size_bytes: int | None = None
    redact_output: bool = True

class Policy(BaseModel):
    name: str
    version: str = "1.0"
    rules: PolicyRules
    on_violation: Literal["block", "log", "redact"] = "block"

from enforcecore.redactor import Redactor, RedactionResult, DetectedEntity
from enforcecore.core.types import RedactionStrategy

# Default: all categories, placeholder strategy
redactor = Redactor()

# Selective categories
redactor = Redactor(categories=["email", "phone"])

# Custom strategy
redactor = Redactor(strategy=RedactionStrategy.HASH)

# Detection only (no modification)
entities: list[DetectedEntity] = redactor.detect("Email john@example.com")
for e in entities:
    print(f"  {e.category}: {e.text!r} at [{e.start}:{e.end}]")

# Full redaction
result: RedactionResult = redactor.redact("Call 555-123-4567 or john@example.com")
print(result.text)           # "Call <PHONE> or <EMAIL>"
print(result.count)          # 2
print(result.was_redacted)   # True
print(result.original_text)  # "Call 555-123-4567 or john@example.com"
print(result.entities)       # [DetectedEntity(category="phone", ...), ...]
print(result.events)         # [RedactionEvent(entity_type="phone", ...), ...]

# policy.yaml
rules:
  pii_redaction:
    enabled: true
    categories: [email, phone, ssn]
    strategy: placeholder

from enforcecore import enforce

@enforce(policy="policy.yaml")
def customer_lookup(query: str) -> str:
    # 'query' is already redacted — PII never reaches your tool
    return db.search(query)

from enforcecore import Auditor, AuditEntry, verify_trail, load_trail

# Create an auditor — writes to a JSONL file
auditor = Auditor(output_path="audit.jsonl")

# Record entries (Merkle-chained automatically)
e1 = auditor.record(
    tool_name="search_web",
    policy_name="default",
    decision="allowed",
    overhead_ms=1.2,
)
e2 = auditor.record(
    tool_name="delete_file",
    policy_name="default",
    decision="blocked",
    violation_type="ToolDeniedError",
    violation_reason="Tool not in allowed list",
)
print(e2.previous_hash == e1.entry_hash)  # True — chain is linked

# Verify trail integrity
result = verify_trail("audit.jsonl")
print(result.is_valid)        # True
print(result.total_entries)   # 2
print(result.chain_intact)    # True
print(result.root_hash)       # SHA-256 of the first entry
print(result.head_hash)       # SHA-256 of the last entry
print(result.error_count)     # 0

# Load entries for analysis
trail: list[AuditEntry] = load_trail("audit.jsonl")
for entry in trail:
    print(f"{entry.tool_name} → {entry.decision}")

from enforcecore import Enforcer, Policy

enforcer = Enforcer(Policy.from_file("policy.yaml"))
result = enforcer.enforce_sync(my_tool, "arg", tool_name="my_tool")
# → Audit entry written automatically to settings.audit_path / "trail.jsonl"

ENFORCECORE_AUDIT_ENABLED=true
ENFORCECORE_AUDIT_PATH=./audit_logs/

# Session 1
a1 = Auditor(output_path="trail.jsonl")
a1.record(tool_name="tool_a", policy_name="p")
a1.record(tool_name="tool_b", policy_name="p")

# Session 2 — chain resumes automatically
a2 = Auditor(output_path="trail.jsonl")
a2.record(tool_name="tool_c", policy_name="p")

result = verify_trail("trail.jsonl")
assert result.is_valid  # True — chain intact across sessions

from enforcecore.guard import ResourceGuard, CostTracker, KillSwitch

# ── CostTracker — thread-safe cumulative cost tracking ──────────────

tracker = CostTracker(budget_usd=10.0)
tracker.record(0.05)            # Record a cost, returns new total
tracker.record(0.10)
print(tracker.total_cost)       # 0.15
print(tracker.budget)           # 10.0
tracker.check_budget()          # Passes — under budget
tracker.reset()                 # Reset to zero

# Per-policy budget check
tracker.check_budget(
    "my_tool", "my_policy",
    per_call_budget=5.0,        # From policy's resource_limits.max_cost_usd
)

# ── KillSwitch — coordinated hard termination ──────────────────────

ks = KillSwitch()
ks.trip("memory exceeded 256MB")
print(ks.is_tripped)            # True
print(ks.reason)                # "memory exceeded 256MB"
ks.check("tool", "policy")     # Raises ResourceLimitError
ks.reset()                      # Re-enable calls

# ── ResourceGuard — time/memory limit enforcement ──────────────────

guard = ResourceGuard(
    cost_tracker=tracker,
    kill_switch=ks,
)

# Sync execution with time limit
result = guard.execute_sync(
    my_func, (arg1,), {"key": "val"},
    max_duration_seconds=30.0,
    max_memory_mb=256,
    tool_name="my_tool",
)

# Async execution with time limit
result = await guard.execute_async(
    my_async_func, (arg1,), {},
    max_duration_seconds=30.0,
)

# Platform support info
info = ResourceGuard.platform_info()
# {"platform": "Darwin", "time_limits": True, "memory_limits": True, ...}

# ── Enforcer integration (automatic) ───────────────────────────────

from enforcecore import Enforcer, Policy

policy = Policy.from_file("policy.yaml")
enforcer = Enforcer(policy)

# Access the guard
guard = enforcer.guard
guard.cost_tracker.total_cost   # Cumulative cost

# Record cost after each call
enforcer.record_cost(2.50)

# Resource limits come from policy YAML:
# rules:
#   resource_limits:
#     max_call_duration_seconds: 30
#     max_memory_mb: 256
#     max_cost_usd: 5.00

from dataclasses import dataclass
from enforcecore.core.types import EnforcementResult

@dataclass
class EnforcementResult:
    decision: str               # "allowed" | "blocked" | "redacted"
    tool_name: str
    policy_name: str
    overhead_ms: float
    redactions_input: int       # Number of PII entities redacted from input
    redactions_output: int      # Number of PII entities redacted from output
    audit_entry_id: str | None  # UUID of the audit entry
    resource_usage: dict        # {"duration_ms": ..., "memory_mb": ...}

class EnforceCoreError(Exception):
    """Base exception for all EnforceCore errors."""

class PolicyError(EnforceCoreError):
    """Errors related to policy loading/validation/evaluation."""

class PolicyLoadError(PolicyError):
    """Policy file not found or unparseable."""

class PolicyValidationError(PolicyError):
    """Policy does not conform to schema."""

class EnforcementViolation(EnforceCoreError):
    """A call was blocked due to policy violation."""
    tool_name: str
    policy_name: str
    reason: str

class ToolDeniedError(EnforcementViolation):
    """Tool is not in the allowed list or is explicitly denied."""

class DomainDeniedError(EnforcementViolation):
    """Network domain is not allowed by policy."""

class CostLimitError(EnforcementViolation):
    """Cumulative cost exceeds budget."""

class ResourceLimitError(EnforcementViolation):
    """Resource limit (time, memory) exceeded."""

class RedactionError(EnforceCoreError):
    """Error during PII redaction (fails closed — call is blocked)."""

class AuditError(EnforceCoreError):
    """Error writing audit entry (fails closed — call is blocked)."""

class HardeningError(EnforceCoreError):
    """Base exception for hardening validation failures (v1.0.6)."""

class InvalidToolNameError(HardeningError):
    """Tool name is empty, too long, or contains invalid characters."""

class InputTooLargeError(HardeningError):
    """Combined input payload exceeds MAX_INPUT_SIZE_BYTES."""

class EnforcementDepthError(HardeningError):
    """Recursive enforcement nesting exceeds MAX_ENFORCEMENT_DEPTH."""

# Validate a policy file
enforcecore validate policies/strict.yaml

# Verify an audit trail
enforcecore verify audit.jsonl

# Run evaluation suite (v1.0.5+)
enforcecore eval --scenarios all --output results/

# Show version and platform info
enforcecore info

ENFORCECORE_DEFAULT_POLICY=policies/default.yaml
ENFORCECORE_AUDIT_PATH=./audit_logs/
ENFORCECORE_AUDIT_ENABLED=true
ENFORCECORE_REDACTION_ENABLED=true
ENFORCECORE_LOG_LEVEL=INFO
ENFORCECORE_COST_BUDGET_USD=100.0
ENFORCECORE_FAIL_OPEN=false  # NEVER set to true in production
ENFORCECORE_DEV_MODE=false   # Set to true/1/yes to suppress fail_open warnings

from enforcecore.integrations import require_package, wrap_with_policy

from enforcecore.integrations.langgraph import enforced_tool

# Drop-in replacement for @tool
@enforced_tool(policy="policy.yaml")
def search(query: str) -> str:
    """Search the web for information."""
    return web_search(query)

# With all options
@enforced_tool(
    policy="policy.yaml",
    tool_name="web_search",
    description="Custom description for LLM",
    args_schema=SearchInput,   # Pydantic model
    return_direct=True,
)
def search(query: str) -> str: ...

# Async support
@enforced_tool(policy="policy.yaml")
async def fetch(url: str) -> str: ...

from enforcecore.integrations.crewai import enforced_tool

@enforced_tool(policy="policy.yaml")
def calculator(expression: str) -> str:
    """Evaluate a math expression."""
    return str(eval(expression))

# Custom name
@enforced_tool(policy="policy.yaml", tool_name="math_calculator")
def calc(expr: str) -> str: ...

from enforcecore.integrations.autogen import enforced_tool

@enforced_tool(policy="policy.yaml", description="Get the weather")
async def get_weather(city: str) -> str:
    return await weather_api.get(city)

# Description from docstring
@enforced_tool(policy="policy.yaml")
def search(query: str) -> str:
    """Search the web for information."""
    return web_search(query)

from enforcecore.integrations import wrap_with_policy

def my_framework_tool(func, policy):
    """Example custom adapter."""
    enforced = wrap_with_policy(func, policy=policy)
    return MyFramework.register_tool(enforced)

from enforcecore.eval import (
    # Types
    ThreatCategory, Severity, ScenarioOutcome,
    Scenario, ScenarioResult, SuiteResult,
    BenchmarkResult, BenchmarkSuite,
    # Runners
    ScenarioRunner, BenchmarkRunner,
    # Helpers
    get_all_scenarios, get_scenarios_by_category,
    # Reports
    generate_suite_report, generate_benchmark_report, generate_report,
)

runner = ScenarioRunner(policy)

result: SuiteResult = runner.run_all(
    category=ThreatCategory.tool_abuse,  # optional filter
    severity=Severity.high,               # optional filter
    tags={"pii"},                         # optional filter
)

result: SuiteResult = runner.run_quick()

result: ScenarioResult = runner.run_scenario(scenario)

scenarios: list[Scenario] = ScenarioRunner.list_scenarios(
    category=ThreatCategory.prompt_injection,
    severity=Severity.critical,
    tags={"sql"},
)

bench = BenchmarkRunner()
suite: BenchmarkSuite = bench.run_all(iterations=1000)

markdown: str = generate_suite_report(suite_result)

markdown: str = generate_benchmark_report(benchmark_suite)

markdown: str = generate_report(suite_result, benchmark_suite)

scenarios: list[Scenario] = get_all_scenarios()

scenarios: list[Scenario] = get_scenarios_by_category(ThreatCategory.tool_abuse)

from enforcecore import (
    # Validation
    validate_tool_name,
    check_input_size,
    deep_redact,
    # Scope tracking
    enter_enforcement,
    exit_enforcement,
    get_enforcement_depth,
    get_enforcement_chain,
    # Dev mode
    is_dev_mode,
    warn_fail_open,
    # Unicode hardening
    normalize_unicode,
    normalize_homoglyphs,
    decode_encoded_pii,
    prepare_for_detection,
    # Exceptions
    HardeningError,
    InvalidToolNameError,
    InputTooLargeError,
    EnforcementDepthError,
)

validate_tool_name(name: str) -> str

check_input_size(
    args: tuple,
    kwargs: dict,
    max_bytes: int = MAX_INPUT_SIZE_BYTES,
) -> None

deep_redact(
    value: Any,
    redact_fn: Callable[[str], str],
    max_depth: int = 10,
) -> Any

from enforcecore import deep_redact

data = {
    "user": {"name": "John Doe", "email": "john@example.com"},
    "notes": ["Contact Jane at jane@test.com"],
}

redacted = deep_redact(data, redactor.redact_string)
# {"user": {"name": "[REDACTED]", "email": "[REDACTED]"}, ...}

enter_enforcement(tool_name: str) -> None

exit_enforcement() -> None

get_enforcement_depth() -> int

get_enforcement_chain() -> list[str]

is_dev_mode() -> bool

warn_fail_open() -> None

normalize_unicode(text: str) -> str

normalize_homoglyphs(text: str) -> str

decode_encoded_pii(text: str) -> str

prepare_for_detection(text: str) -> str

entries = load_trail(path, max_entries=100)  # Returns last 100 entries

from enforcecore import on_pre_call, on_post_call, on_violation, on_redaction

@on_pre_call
def log_calls(ctx: HookContext):
    print(f"Calling {ctx.tool_name}")

@on_post_call
def measure_latency(ctx: HookContext):
    metrics.record(ctx.tool_name, ctx.duration_ms)

@on_violation
def alert_on_block(ctx: ViolationHookContext):
    slack.post(f"BLOCKED: {ctx.tool_name} — {ctx.violation_reason}")

@on_redaction
def track_redactions(ctx: RedactionHookContext):
    counter.increment(f"pii.{ctx.direction}.{ctx.category}", ctx.redaction_count)

from enforcecore import HookRegistry

registry = HookRegistry.global_registry()
registry.add_pre_call(my_hook_fn)
registry.add_violation(my_alert_fn)

# Instance isolation (for testing)
isolated = HookRegistry()
isolated.add_pre_call(test_hook)

# Count & clear
print(registry.total_count)
registry.clear()

@on_pre_call
def rate_limiter(ctx: HookContext):
    if is_rate_limited(ctx.tool_name):
        ctx.abort = True
        ctx.abort_reason = "Rate limit exceeded"

@on_pre_call
async def async_audit(ctx: HookContext):
    await audit_service.log(ctx.tool_name)

from enforcecore import PatternRegistry

# Register a domain-specific pattern
PatternRegistry.register(
    "employee_id",
    r"EMP-\d{6}",
    placeholder="<EMPLOYEE_ID>",
    mask="EMP-******",
)

# With validation to reduce false positives
PatternRegistry.register(
    "medical_record",
    r"MRN-\d{8}",
    validator=lambda match: match.startswith("MRN-"),
)

# Query
PatternRegistry.categories()   # ["employee_id", "medical_record"]
PatternRegistry.count()        # 2

# Unregister
PatternRegistry.unregister("employee_id")

# Instance isolation (for testing)
isolated = PatternRegistry()
isolated.add("test_pattern", r"TEST-\d+")

from enforcecore import SecretScanner, DetectedSecret

scanner = SecretScanner()
secrets = scanner.detect("My key is AKIAIOSFODNN7EXAMPLE")
# [DetectedSecret(category="aws_access_key", start=10, end=30, text="AKIA...")]

# Limit to specific categories
scanner = SecretScanner(categories=["github_token", "private_key"])

# Quick audit
counts = scanner.scan_and_report(text)
# {"aws_access_key": 1, "github_token": 2}

from enforcecore import (
    AuditBackend, JsonlBackend, NullBackend,
    CallbackBackend, MultiBackend, Auditor,
)

# Default JSONL backend
backend = JsonlBackend("audit/trail.jsonl")

# Null backend (testing)
backend = NullBackend()

# Callback backend (send to custom pipeline)
entries = []
backend = CallbackBackend(entries.append)

# Error handling
backend = CallbackBackend(
    send_to_siem,
    on_error=lambda exc, entry: logger.error(f"SIEM error: {exc}"),
)

# Fan-out to multiple backends
backend = MultiBackend([
    JsonlBackend("audit.jsonl"),
    CallbackBackend(send_to_siem),
])

# Use with Auditor
auditor = Auditor(backend=backend)

# Custom backend
class S3Backend(AuditBackend):
    def write(self, entry_dict: dict) -> None:
        s3.put_object(Body=json.dumps(entry_dict), ...)

    def close(self) -> None:
        pass  # flush buffers if any

from enforcecore import AuditEntry

@classmethod
def create(
    cls,
    policy_name: str,
    policy_version: str,
    tool_name: str,
    decision: str,
    call_duration_ms: float,
    enforcement_overhead_ms: float,
    input_redactions: int = 0,
    output_redactions: int = 0,
    **kwargs: Any,
) -> AuditEntry

def to_dict(self) -> dict[str, Any]

from enforcecore.auditstore import AuditStore

def __init__(self, backend: AuditBackend, verify_on_read: bool = True) -> None

def record(self, **kwargs: Any) -> AuditEntry

def get_entry(self, entry_id: str) -> AuditEntry | None

def list_entries(
    self,
    policy_name: str | None = None,
    tool_name: str | None = None,
    start_time: datetime | None = None,
    end_time: datetime | None = None,
    decision: str | None = None,
    limit: int = 1000,
    offset: int = 0,
) -> list[AuditEntry]

def verify_chain(self, start_index: int = 0, end_index: int | None = None) -> bool

def verify_entry(self, entry: AuditEntry) -> bool

def export(self, format: str = "jsonl") -> str

from enforcecore.auditstore.backends import AuditBackend

@abstractmethod
def record(self, entry: AuditEntry) -> AuditEntry
# Must compute and set merkle_hash, parent_hash, chain_index.

@abstractmethod
def get_entry(self, entry_id: str) -> AuditEntry | None

@abstractmethod
def list_entries(self, ...) -> list[AuditEntry]

@abstractmethod
def get_chain_tail(self) -> AuditEntry | None

@abstractmethod
def verify_chain(self, start_index: int = 0, end_index: int | None = None) -> bool

@abstractmethod
def verify_entry(self, entry: AuditEntry) -> bool

@abstractmethod
def export(self, format: str = "jsonl") -> str

# Concrete helper — provided for all subclasses:
def _compute_merkle_hash(self, entry: AuditEntry, parent_hash: str | None = None) -> str
# SHA-256 over {entry_id, timestamp, policy_name, tool_name, decision,
#               violation_type, parent_hash}. Genesis parent_hash = "0" * 64.

from enforcecore.auditstore.backends import JSONLBackend

backend = JSONLBackend("audit_logs/trail.jsonl")

def __init__(self, path: str = "audit_logs/trail.jsonl")
# Parent directories created automatically (mkdir -p).

from enforcecore.auditstore.backends import SQLiteBackend

backend = SQLiteBackend("audit.db")

def __init__(self, db_path: str = "audit.db", verify_merkle: bool = True)

from enforcecore.auditstore.backends import PostgreSQLBackend
# Requires: pip install enforcecore[postgres]

backend = PostgreSQLBackend(
    host="localhost",
    database="enforcecore",
    user="postgres",
    password="secret",
    port=5432,
    verify_merkle=True,
    pool_size=5,
    ssl_mode="prefer",  # "disable" | "allow" | "prefer" | "require"
)

from enforcecore import AuditStoreBackendAdapter

adapter = AuditStoreBackendAdapter(backend)
auditor = Auditor(backend=adapter)

def __init__(self, auditstore_backend: AuditBackend) -> None

def write(self, entry_dict: dict[str, Any]) -> None
# Raises AuditError on failure.

def close(self) -> None

from enforcecore.auditstore.queries.eu_ai_act import EUAIActQueries

queries = EUAIActQueries(store)

from enforcecore.auditstore.reports import ReportGenerator

generator = ReportGenerator(store)
report = generator.generate_eu_ai_act_report(
    organization="Acme Corp",
    period="Q1 2026",    # or "Jan 2026", "January 2026"
    format="html",       # or "json"
)
report.save("compliance_report.html")

def render(self) -> str
def save(self, filename: str) -> None

from enforcecore import SandboxStrategy

SandboxStrategy.NONE        # In-process execution (no isolation)
SandboxStrategy.SUBPROCESS  # Isolated subprocess with resource limits

from enforcecore import SandboxConfig

config = SandboxConfig(
    strategy=SandboxStrategy.SUBPROCESS,
    timeout_seconds=30.0,
    max_memory_mb=256,
    allowed_env_vars=["PATH", "HOME"],
    capture_output=True,
)

from enforcecore import SubprocessSandbox

sandbox = SubprocessSandbox(config)

# Synchronous execution
result = sandbox.run(my_tool, *args, tool_name="my_tool", **kwargs)

# Asynchronous execution
result = await sandbox.run_async(my_tool, *args, tool_name="my_tool", **kwargs)

def __init__(self, config: SandboxConfig) -> None

def run(
    self,
    func: Callable[..., Any],
    *args: Any,
    tool_name: str = "",
    policy_name: str = "",
    **kwargs: Any,
) -> Any

async def run_async(
    self,
    func: Callable[..., Any],
    *args: Any,
    tool_name: str = "",
    policy_name: str = "",
    **kwargs: Any,
) -> Any

from enforcecore import SandboxTimeoutError, SandboxMemoryError, SandboxViolationError

# SandboxTimeoutError   — subprocess exceeded timeout_seconds
# SandboxMemoryError    — subprocess exceeded max_memory_mb
# SandboxViolationError — other sandbox failures (non-picklable result, etc.)

from enforcecore.sandbox.runner import platform_info

info = platform_info()
# Returns dict: {platform, subprocess_isolation, resource_limits,
#                memory_limits, cpu_time_limits, env_restriction,
#                wasm_sandbox, python_version}

from enforcecore import NERBackend
# Requires: pip install enforcecore[ner]
# Requires: python -m spacy download en_core_web_lg

ner = NERBackend(threshold=0.8, language="en")
spans = ner.analyze("Call John at john@acme.com")
# -> [(9, 13, "person_name", 0.95), (17, 29, "email", 0.99)]

def __init__(self, threshold: float = 0.8, language: str = "en") -> None
# Raises ImportError if presidio-analyzer not installed or spaCy model unavailable.

def analyze(
    self,
    text: str,
    categories: set[str] | None = None,
    *,
    threshold: float | None = None,
) -> list[tuple[int, int, str, float]]
# Returns [(start, end, category, score), ...] sorted by start position.
# categories=None -> all 17 supported entity types.

@property
def threshold(self) -> float

@property
def language(self) -> str

from enforcecore import is_ner_available

if is_ner_available():
    ner = NERBackend()

from enforcecore import SensitivityLabel

SensitivityLabel.PUBLIC        # 0 — openly shareable
SensitivityLabel.INTERNAL      # 1 — company-internal only
SensitivityLabel.CONFIDENTIAL  # 2 — restricted to authorised roles
SensitivityLabel.RESTRICTED    # 3 — highest sensitivity, minimum distribution

from enforcecore import SensitivityEnforcer, SensitivityLabel

enforcer = SensitivityEnforcer(
    tool_clearance=SensitivityLabel.INTERNAL,
    field_labels={
        "body": SensitivityLabel.CONFIDENTIAL,
        "subject": SensitivityLabel.INTERNAL,
        "to": SensitivityLabel.PUBLIC,
    },
    default_field_sensitivity=SensitivityLabel.PUBLIC,
)

violations = enforcer.check_kwargs({"body": "...", "to": "..."})
enforcer.raise_if_violated(violations, tool_name="send_email")

def __init__(
    self,
    tool_clearance: SensitivityLabel | str,
    field_labels: dict[str, SensitivityLabel | str] | None = None,
    *,
    default_field_sensitivity: SensitivityLabel | str = SensitivityLabel.PUBLIC,
) -> None

def check(self) -> list[SensitivityViolation]

def check_kwargs(
    self,
    kwargs: dict[str, Any],
    *,
    override_labels: dict[str, SensitivityLabel | str] | None = None,
) -> list[SensitivityViolation]

def raise_if_violated(
    self,
    violations: list[SensitivityViolation],
    *,
    tool_name: str = "",
    policy_name: str = "",
) -> None

@property
def tool_clearance(self) -> SensitivityLabel

@property
def field_labels(self) -> dict[str, SensitivityLabel]

from enforcecore import check_tool_schema_sensitivity, SensitivityLabel

tool_schema = {
    "name": "send_email",
    "clearance": "internal",
    "parameters": {
        "to":   {"type": "string", "sensitivity": "low"},
        "body": {"type": "string", "sensitivity": "high"},
    },
}
violations = check_tool_schema_sensitivity(tool_schema, SensitivityLabel.INTERNAL)

def check_tool_schema_sensitivity(
    tool_schema: dict[str, Any],
    policy_clearance: SensitivityLabel | str,
) -> list[SensitivityViolation]
# Effective clearance = min(schema["clearance"], policy_clearance).

@dataclass(frozen=True)
class SensitivityViolation:
    field_name: str
    field_sensitivity: SensitivityLabel
    tool_clearance: SensitivityLabel

from enforcecore import sensitivity_level, SensitivityLabel

level = sensitivity_level("confidential")
# -> SensitivityLabel.CONFIDENTIAL

sensitivity_labels:
  enabled: true
  default_clearance: internal
  enforce: true
  fallback: redact

from enforcecore import EnforceCoreInstrumentor
# Requires: pip install enforcecore[otel]

instrumentor = EnforceCoreInstrumentor()
instrumentor.instrument(
    tracer_provider=my_provider,  # optional; falls back to global OTel provider
    meter_provider=my_meter,      # optional
)

def instrument(
    self,
    *,
    tracer_provider: Any = None,
    meter_provider: Any = None,
) -> None

def uninstrument(self) -> None

@property
def metrics(self) -> EnforceCoreMetrics

@property
def is_instrumented(self) -> bool

from enforcecore import EnforceCorePrometheusExporter
# Requires: pip install enforcecore[prometheus]

exporter = EnforceCorePrometheusExporter()
exporter.instrument()
exporter.start_http_server(port=9090)

def __init__(self, *, registry: Any = None) -> None
# registry: prometheus_client.CollectorRegistry or None (uses global registry).
# Operates in no-op mode if prometheus_client not installed — never raises on init.

def instrument(self) -> None
def uninstrument(self) -> None
def start_http_server(self, port: int = 9090, addr: str = "0.0.0.0") -> None

@property
def is_instrumented(self) -> bool

@property
def is_available(self) -> bool

from enforcecore.telemetry import AuditLogExporter

exporter = AuditLogExporter(
    sink=make_splunk_hec_sink("https://splunk.acme.com:8088", token="xxx"),
)

from enforcecore.telemetry import make_splunk_hec_sink, make_elastic_sink

splunk_sink = make_splunk_hec_sink(url, token, *, index=None, source=None)
elastic_sink = make_elastic_sink(url, index, *, api_key=None, username=None, password=None)

from enforcecore import MultiTenantEnforcer

mt = MultiTenantEnforcer(default_policy="base_policy.yaml")
mt.register("team_a", "team_a_policy.yaml")
mt.register("team_b", "team_b_policy.yaml")

def __init__(self, default_policy: Policy | str | Path | None = None) -> None

def register(self, tenant_id: str, policy: Policy | str | Path) -> None
# Raises ValueError if tenant_id is empty.

def unregister(self, tenant_id: str) -> None
# Raises KeyError if tenant not registered.

@property
def tenants(self) -> list[str]   # sorted

def get_enforcer(self, tenant_id: str) -> Enforcer
# Raises KeyError if not registered and no default policy.

def enforce_sync(
    self,
    tenant_id: str,
    func: Callable[..., T],
    /,
    *args: Any,
    tool_name: str | None = None,
    **kwargs: Any,
) -> T

async def enforce_async(
    self,
    tenant_id: str,
    func: Callable[..., Any],
    /,
    *args: Any,
    tool_name: str | None = None,
    **kwargs: Any,
) -> Any

def __len__(self) -> int
def __contains__(self, tenant_id: object) -> bool

# agent_deployer.yaml
extends: team_dev.yaml
tools:
  allowed: [search, calculate, deploy_service]
  context:
    environment: [staging]

from enforcecore import PolicyServerClient

client = PolicyServerClient(
    url="https://policy.acme.com/agents/chatbot-v2",
    token=os.environ["POLICY_SERVER_TOKEN"],
    cache_ttl=300,          # seconds; 0 disables caching
    verify_signature=True,  # HMAC-SHA256 verification
)
enforcer = Enforcer.from_server(client)

# Or directly:
enforcer = Enforcer.from_server(
    "https://policy.acme.com/agents/chatbot-v2",
    token=os.environ["POLICY_SERVER_TOKEN"],
    cache_ttl=300,
)

def __init__(
    self,
    url: str,
    token: str,
    *,
    cache_ttl: int = 300,
    verify_signature: bool = True,
) -> None
# Raises ValueError if url/token empty or cache_ttl < 0.

def get_policy(self) -> Policy
# Returns cached policy if fresh (cache_ttl > 0 and not expired).
# On cache miss: fetches from server, verifies HMAC signature (if header present).
# Stale-on-error: returns cached policy if server unreachable and cache exists.
# Raises PolicyServerError if server unreachable and no cache.

def invalidate(self) -> None
# Forces re-fetch on next get_policy().

@property
def url(self) -> str

@property
def cache_ttl(self) -> int

@property
def policy_version(self) -> str | None  # from X-Policy-Version response header

from enforcecore import ComplianceReporter, ComplianceFormat, CompliancePeriod

reporter = ComplianceReporter()
report = reporter.export(
    ComplianceFormat.EU_AI_ACT,
    CompliancePeriod.from_label("2026-Q4"),
)

def export(
    self,
    format: ComplianceFormat,
    period: CompliancePeriod,
    *,
    organization: str = "",
) -> ComplianceReport

def export_json(self, ...) -> str
def export_html(self, ...) -> str
def send_webhook(self, url: str, ...) -> None
# Vanta / Drata webhook integration.

ComplianceFormat.EU_AI_ACT  # EU AI Act Articles 9, 13, 14, 52
ComplianceFormat.SOC2       # SOC2 CC6, CC7, CC8, CC9
ComplianceFormat.GDPR       # GDPR Article 30 (records of processing)

period = CompliancePeriod.from_label("2026-Q4")  # quarterly
period = CompliancePeriod.from_label("2026-H1")  # half-year
period = CompliancePeriod.from_label("2026")     # annual

@dataclass
class ComplianceReport:
    title: str
    format: ComplianceFormat
    period: CompliancePeriod
    content: str           # rendered report body
    score: float           # 0.0 – 1.0 compliance score
    narratives: list[str]  # human-readable compliance statements
    statistics: dict[str, Any]
    generated_at: datetime

enforcecore audit export --format eu-ai-act --period 2026-Q4 --output report.json
enforcecore audit export --format soc2 --period 2026-H1 --output soc2.html

from enforcecore import GuardPlugin, GuardResult

class MyGuard(GuardPlugin):

    @property
    def name(self) -> str:
        return "my-guard"

    @property
    def version(self) -> str:
        return "1.0.0"

    def check(
        self,
        tool_name: str,
        args: tuple[Any, ...],
        kwargs: dict[str, Any],
    ) -> GuardResult:
        if tool_name == "dangerous_tool":
            return GuardResult(allowed=False, reason="blocked by policy")
        return GuardResult(allowed=True)

[project.entry-points."enforcecore.guards"]
my-guard = "my_package.guards:MyGuard"

from enforcecore import RedactorPlugin, RedactResult

class MyRedactor(RedactorPlugin):

    @property
    def name(self) -> str:
        return "my-redactor"

    @property
    def version(self) -> str:
        return "1.0.0"

    @property
    def categories(self) -> list[str]:
        return ["internal_id"]

    def redact(self, text: str) -> RedactResult:
        cleaned, count = _scrub_internal_ids(text)
        return RedactResult(text=cleaned, count=count)

from enforcecore import AuditBackendPlugin

class MyBackend(AuditBackendPlugin):

    @property
    def name(self) -> str:
        return "my-backend"

    @property
    def version(self) -> str:
        return "1.0.0"

    def record(self, entry: dict[str, Any]) -> None:
        # entry matches AuditEntry.to_dict() schema
        my_db.insert(entry)

    def flush(self) -> None:
        my_db.flush()

@dataclass(frozen=True)
class GuardResult:
    allowed: bool
    reason: str = ""

@dataclass(frozen=True)
class RedactResult:
    text: str
    count: int = 0

@dataclass(frozen=True)
class PluginInfo:
    name: str
    version: str
    kind: str         # "guard" | "redactor" | "audit_backend"
    package: str = ""
    description: str = ""

from enforcecore import PluginManager

manager = PluginManager()

# Discover without loading (metadata only):
infos = manager.discover()

# Load all installed plugins:
count = manager.load_all(ignore_errors=False)
# Raises PluginLoadError if any plugin fails and ignore_errors=False.

# Load a single plugin by name:
manager.load("my-guard")

# Access loaded plugins:
manager.guards          # list[GuardPlugin]
manager.redactors       # list[RedactorPlugin]
manager.audit_backends  # list[AuditBackendPlugin]

def discover(self) -> list[PluginInfo]
def load_all(self, *, ignore_errors: bool = False) -> int
def load(self, name: str) -> None
# Raises PluginLoadError if name not found or instantiation fails.

@property
def guards(self) -> list[GuardPlugin]

@property
def redactors(self) -> list[RedactorPlugin]

@property
def audit_backends(self) -> list[AuditBackendPlugin]

from enforcecore import PluginLoadError

enforcecore plugin list           # Discover all installed plugins
enforcecore plugin info <name>    # Load and inspect a specific plugin

from enforcecore import stream_enforce, Policy
from collections.abc import AsyncGenerator

policy = Policy.from_file("policies/production.yaml")

async def run(llm_stream: AsyncGenerator[str, None]) -> None:
    result_holder: list[StreamEnforcementResult] = []
    async with stream_enforce(
        llm_stream,
        policy=policy,
        tool_name="chatbot_reply",
        redact=True,
        redaction_strategy="placeholder",   # placeholder | mask | hash | remove
        lookahead=64,                        # chars buffered for boundary detection
        result_out=result_holder,
    ) as safe:
        async for token in safe:
            print(token, end="", flush=True)

    result = result_holder[0]
    print(f"\nRedacted {result.total_redactions} PII entities in {result.stream_duration_ms:.1f}ms")

from enforcecore import StreamingRedactor

sr = StreamingRedactor(strategy="placeholder", lookahead=64)

async for raw_token in llm_stream:
    safe_token, events = sr.push(raw_token)
    yield safe_token

remainder, events = sr.flush()  # flush buffered lookahead at stream end
yield remainder

@dataclass
class StreamEnforcementResult:
    stream_id: str           # UUID
    policy_name: str
    tool_name: str
    decision: str            # "allowed" | "blocked"
    violation_reason: str | None
    tokens_total: int
    tokens_redacted: int
    total_redactions: int
    overhead_ms: float
    stream_duration_ms: float
    audit_entry_id: str | None

@dataclass
class StreamAuditEntry:
    entry_id: str            # UUID
    stream_id: str
    timestamp: str           # ISO-8601
    policy_name: str
    tool_name: str
    decision: str
    violation_reason: str | None
    tokens_emitted: int
    tokens_redacted: int
    total_redactions: int
    redaction_events: list[StreamRedactionEvent]
    overhead_ms: float
    stream_duration_ms: float

class StreamingViolation(Exception):
    reason: str
    stream_id: str
    result: StreamEnforcementResult

from enforcecore.streaming.adapters import EnforceCoreStreamingCallback

callback = EnforceCoreStreamingCallback(policy=policy, redact=True)

# Wrap the LLM's token stream:
async for token in callback.on_llm_new_token_stream(llm_token_generator):
    print(token, end="")

result = callback.last_result  # StreamEnforcementResult

from enforcecore.streaming.adapters import autogen_stream_enforce

async for token in autogen_stream_enforce(agent_stream, policy=policy):
    print(token, end="")

from enforcecore.streaming.adapters import langgraph_stream_enforce

async with langgraph_stream_enforce(
    graph.astream(inputs),
    policy=policy,
    token_extractor=lambda e: e.get("content") if isinstance(e, dict) else None,
) as safe_events:
    async for event in safe_events:
        process(event)