Skip to content

Policy Enforcement

Policies are runtime constraints that enforce governance, compliance, and security requirements across workflow execution.

What are Policies?

Policies are declarative rules that:

  • Prevent unsafe tool compositions
  • Enforce data privacy requirements
  • Validate schemas and constraints
  • Audit execution for compliance

Policies are enforced before, during, and after execution—not as code review checklists.

Built-in Policies

1. privacy_strict

Prevents PII leakage by requiring redaction before downstream tools.

Rules: - ✅ PII redaction must occur before summarization - ✅ Unredacted text cannot leave the mesh - ❌ Blocks workflows that skip redaction

Example: 

{
  "workflow_id": "safe_summary",
  "steps": [
    { "tool_name": "pii_redact", ... },
    { "tool_name": "summarize", ... }
  ],
  "policies": ["privacy_strict"]
}

2. schema_strict

Validates all inputs and outputs against JSON schemas.

Rules: - ✅ Tool inputs must match input_schema - ✅ Tool outputs must match output_schema - ❌ Blocks execution if validation fails

Example: 

{
  "workflow_id": "validated_workflow",
  "steps": [
    {
      "tool_name": "validate_schema",
      "tool_args": {
        "data": "{{input.feedback}}",
        "schema": "feedback_v1"
      }
    }
  ],
  "policies": ["schema_strict"]
}

3. offline_capable

Ensures workflows can run without internet.

Rules: - ✅ All tools must be LAN-available - ✅ No external API dependencies - ❌ Blocks workflows requiring cloud services

4. audit_full

Enables complete execution trace logging.

Rules: - ✅ Log all tool invocations - ✅ Record execution paths - ✅ Generate cryptographic evidence - ✅ Store audit trail in audit/audit.log

Policy Lifecycle

Before Execution

Policy Check Phase:

  1. Parse workflow definition
  2. Validate policy compatibility
  3. Check required tools exist
  4. Verify capability tokens

If any check fails, execution is rejected before starting.

During Execution

Runtime Enforcement:

  • Schema validation on every tool input/output
  • PII detection and blocking
  • Capability token verification
  • Execution path recording

After Execution

Audit Generation:

  • Execution trace with timestamps
  • Cryptographic signatures
  • Result hashes for tamper detection
  • Decision log (chain-of-thought)

Custom Policies

Define custom policies in runtime/policies/:

# runtime/policies/custom_policy.py

class CustomPolicy(Policy):
    """Require data encryption before external transmission."""

    def check_workflow(self, workflow: Workflow) -> PolicyResult:
        # Find steps that transmit data externally
        external_steps = [
            s for s in workflow.steps
            if s.target.kind == "external"
        ]

        # Ensure encryption occurs first
        for step in external_steps:
            if not self._has_prior_encryption(workflow, step):
                return PolicyResult(
                    ok=False,
                    error="External transmission without encryption"
                )

        return PolicyResult(ok=True)

    def _has_prior_encryption(self, workflow, step):
        # Check if an encryption step precedes this step
        ...

Registering Custom Policies

from runtime.policies import register_policy
from runtime.policies.custom_policy import CustomPolicy

register_policy("require_encryption", CustomPolicy())

Using Custom Policies

{
  "workflow_id": "secure_transmission",
  "steps": [
    { "tool_name": "encrypt_data", ... },
    { "tool_name": "send_external", ... }
  ],
  "policies": ["require_encryption"]
}

Policy Violations

Example: PII Leak Attempt

Workflow (without redaction): 

{
  "steps": [
    { "tool_name": "summarize", "tool_args": { "text": "Email: john@example.com" } }
  ],
  "policies": ["privacy_strict"]
}

Result: 

{
  "ok": false,
  "error": "Policy violation: privacy_strict",
  "details": "Unredacted PII detected in summarization step",
  "violated_at": "step:summarize"
}

Example: Schema Validation Failure

Workflow (invalid input): 

{
  "steps": [
    {
      "tool_name": "validate_schema",
      "tool_args": { "data": 12345 }  // Should be object
    }
  ],
  "policies": ["schema_strict"]
}

Result: 

{
  "ok": false,
  "error": "Schema validation failed",
  "details": "Expected object, got number",
  "step_id": "validate"
}

Policy Composition

Policies can be combined:

{
  "policies": [
    "privacy_strict",
    "schema_strict",
    "offline_capable",
    "audit_full"
  ]
}

All policies must pass for execution to proceed.

Audit Trail Example

With audit_full policy enabled:

// audit/audit.log (JSONL)
{
  "timestamp": "2024-01-27T10:30:00Z",
  "trace_id": "trace-xyz",
  "workflow_id": "feedback_safe_summary",
  "step_id": "validate",
  "tool_name": "validate_schema",
  "input": { "data": "Contact me at john@example.com" },
  "output": { "valid": true },
  "node_id": "local:abc123",
  "signature": "ed25519:...",
  "duration_ms": 42
}
{
  "timestamp": "2024-01-27T10:30:01Z",
  "trace_id": "trace-xyz",
  "workflow_id": "feedback_safe_summary",
  "step_id": "redact",
  "tool_name": "pii_redact",
  "input": { "text": "Contact me at john@example.com" },
  "output": { "redacted_text": "Contact me at [REDACTED]" },
  "node_id": "local:abc123",
  "signature": "ed25519:...",
  "duration_ms": 28
}

Best Practices

1. Default to Strict Policies

Start with comprehensive policies:

{
  "policies": [
    "privacy_strict",
    "schema_strict",
    "audit_full"
  ]
}

Relax only when necessary.

2. Test Policy Violations

Ensure policies actually block unsafe workflows:

python tests/test_policies.py

3. Review Audit Logs

Regularly review audit/audit.log for compliance:

cat audit/audit.log | jq 'select(.workflow_id == "feedback_safe_summary")'

4. Document Policy Requirements

Clearly state which policies apply to each workflow:

## Workflow: HIPAA Field Report

**Required Policies**:
- `privacy_strict` - HIPAA compliance
- `schema_strict` - Data validation
- `audit_full` - Legal audit trail

Next Steps