Runtime Enforcement for AI: Practical Compliance-as-Code Patterns for Developers

Why runtime enforcement matters now

The EU AI Act and similar regulations push responsibilities for AI behavior into production: it's not enough to review models during development. Systems must enforce rules at runtime, create auditable evidence, and be able to block or alter outputs that violate policy. This article gives practical patterns you can apply today to build runtime enforcement that scales with latency, observability, and developer ergonomics in mind.

Core primitives of runtime enforcement

• Policy evaluation — centralized or embedded policy engine that decides allow/deny/modify for requests and model outputs.
• Input/output filters — sanitizers and classifiers applied before/after model calls.
• Context enrichment — attach provenance, user attributes, and risk scores to decisions.
• Gating and canarying — progressive rollout with holdouts and human-in-the-loop review for high-risk outputs.
• Observability & audit logs — immutable records of inputs, policy decisions, and model responses for compliance and debugging.

Pick a policy engine: centralized vs. embedded

Two common approaches:

1. Centralized policy service — run Open Policy Agent (OPA) or a policy API as a sidecar. Benefits: single source of truth, easy updates. Drawbacks: network latency and an additional operational component.
2. Embedded policy checks — compile policies into your service (WASM, libraries). Benefits: lower latency, simpler call graph. Drawbacks: redeploys for policy changes and potential drift.

Example Rego policy (high level)

package ai.runtime
 
# Deny output that contains a banned phrase or flagged PII label
deny[msg] {
  output := input.output
  banned := ["classified", "secret"]
  some i
  contains(output, banned[i])
  msg = sprintf("banned phrase detected: %v", [banned[i]])
}
 
contains(s, sub) {
  indexof(s, sub) >= 0
}

Node.js Express middleware: call a policy service

This pattern shows intercepting a user request, calling your model, then asking a policy service whether the model output is allowed. Keep the model call and policy call parallel when possible to reduce tail latency.

// Express middleware (simplified)
const express = require('express');
const fetch = require('node-fetch');
const app = express();
 
app.use(express.json());
 
app.post('/chat', async (req, res) => {
  const user = req.body.user;
  const prompt = req.body.prompt;
 
  // 1) call the model (async)
  const modelPromise = fetch('http://model:8000/generate', {
    method: 'POST',
    headers: {'content-type': 'application/json'},
    body: JSON.stringify({prompt, user})
  }).then(r => r.json());
 
  // 2) run policy check in parallel
  const policyPromise = modelPromise.then(modelResp =>
    fetch('http://policy:8181/v1/data/ai/runtime/deny', { // OPA REST API
      method: 'POST',
      headers: {'content-type': 'application/json'},
      body: JSON.stringify({input: {user, prompt, output: modelResp.text}})
    }).then(r => r.json())
  );
 
  const [modelResp, policyResp] = await Promise.all([modelPromise, policyPromise]);
 
  if (policyResp && policyResp.result) {
    // policy returned a deny message
    return res.status(403).json({error: 'output blocked', reason: policyResp.result});
  }
 
  // attach audit metadata and return
  // store audit event asynchronously
  fetch('http://events:9000/audit', {method: 'POST', body: JSON.stringify({user, prompt, output: modelResp.text, policy: policyResp})});
 
  res.json({text: modelResp.text});
});

Python FastAPI dependency: post-check the model output

Use a dependency to enforce policy across routes. This example assumes a synchronous policy HTTP API (e.g., OPA).

from fastapi import FastAPI, HTTPException, Depends
import requests
 
app = FastAPI()
 
def policy_check(user, prompt, output):
    resp = requests.post('http://policy:8181/v1/data/ai/runtime/deny', json={'input': {'user': user, 'prompt': prompt, 'output': output}})
    data = resp.json()
    if data.get('result'):
        raise HTTPException(status_code=403, detail={'blocked': data['result']})
 
@app.post('/generate')
def generate(payload: dict):
    user = payload.get('user')
    prompt = payload.get('prompt')
    model_resp = requests.post('http://model:8000/generate', json={'prompt': prompt}).json()
    output = model_resp.get('text')
    policy_check(user, prompt, output)
    # log audit asynchronously here
    return {'text': output}

Practical implementation checklist

• Start with a small set of enforcements (PII blocking, hate speech, safety-critical instructions).
• Choose centralized (OPA) or embedded policies based on latency budget.
• Instrument model calls with a unique request ID and store inputs, outputs, policy decisions, and versioned model metadata.
• Use differential logging: sample full inputs for lower-risk traffic, full logs for high-risk transactions.
• Implement human-in-the-loop review workflows for blocked outputs and an appeals pipeline.
• Add tests: unit tests for policy logic and integration tests that simulate model outputs.

Tradeoffs and operational concerns

• Latency vs. correctness: synchronous policy checks add latency. Mitigate with parallel calls, caching policy decisions, or local WASM policies.
• False positives: strict policies can unnecessarily block legitimate outputs. Provide override paths and clear human review UIs.
• Policy complexity: complex rules are harder to reason about. Keep policies modular and well-tested.
• Auditability: store enough data to reconstruct decisions but respect data minimization and privacy rules.
• Rollout complexity: gating and canarying require feature flags and observability to detect regressions quickly.

Integration with CI/CD and testing

Treat policies as code: keep them in the same repo or a mirrored repo, add unit tests and policy linting, and include policy validation in your CI pipeline. Run integration tests that exercise policy endpoints with mocked model outputs to ensure expected allow/deny behavior before production rollouts.

Further resources

• Open Policy Agent (OPA) — widely used policy engine useful as a centralized decision point.
• Context and regulatory push: Compliance as Code (DEV)

Conclusion

Runtime enforcement is becoming a practical requirement for production AI systems. Start small: implement a few clear, testable policies, wire a policy evaluation step into your request flow, and add audit logs. Balance latency, accuracy, and maintainability by choosing the right policy deployment model (centralized vs. embedded) and by adding human review and sampling for edge cases. With policies treated as code and proper observability, you can move quickly while keeping your system auditable and compliant.