Serverless Deployment

The Python examples on this page use tork-governance, whose only entry class is Tork. That is the right choice for a function that must decide inside its own runtime. govern() is synchronous — do not await it.

tork-governance makes the governance decision on your machine. PII detection, redaction and the returned verdict are computed on-device. With no API key it makes zero network calls: prompts, completions and detected PII values never leave the machine, and nothing appears in your dashboard.

Supply an API key and the SDK additionally reports a metadata-only attestation of each decision — the action taken, PII type labels and counts, a risk classification, policy labels and a salted fingerprint. It never sends input text, output text or PII values. The decision itself is still made on-device and is never delayed or changed by reporting. Those attestations appear in your dashboard and are included in the daily on-chain anchor — each one recorded as a client attestation (capture_mode=edge, attested_by=client), a claim Tork recorded but did not execute and cannot independently verify. Requires tork-governance 0.24.0+ (PyPI) or 0.11.0+ (npm).

A decision reported by an on-device SDK is recorded as a client attestation (capture_mode=edge, attested_by=client): a claim Tork recorded but did not execute and cannot independently verify. A decision made by @torknetwork/sdk is recorded as capture_mode=cloud, attested_by=tork — Tork made that call itself. Both are equally immutable once anchored; they differ in what is immutable. A client attestation freezes your claim. A server-governed receipt freezes Tork's own decision.

"""
AWS Lambda handler for Tork AI Governance.

Provides serverless governance evaluation for AI requests.
"""

import json
import os
from typing import Any

from tork_governance import Tork, detect_pii, redact_pii


# Initialize outside handler for warm start reuse
# This persists across invocations in the same container
tork = None


def initialize():
    """Initialize Tork (called once per container)."""
    global tork

    if tork is None:
        # With no api_key this is fully local: zero network calls, and the
        # decision is computed inside this Lambda. Passing api_key additionally
        # POSTs a metadata-only client attestation on a background thread.
        tork = Tork(api_key=os.environ.get("TORK_API_KEY"))

        # Pre-warm the detector so the first real request isn't the one
        # that pays for pattern compilation.
        detect_pii("warmup")


def lambda_handler(event: dict, context: Any) -> dict:
    """
    Main Lambda handler for governance evaluation.

    Args:
        event: Lambda event (API Gateway or direct invocation)
        context: Lambda context object

    Returns:
        API Gateway response or direct result
    """
    initialize()

    # Parse request body
    if "body" in event:
        # API Gateway invocation
        try:
            body = json.loads(event["body"]) if isinstance(event["body"], str) else event["body"]
        except json.JSONDecodeError:
            return {
                "statusCode": 400,
                "headers": {"Content-Type": "application/json"},
                "body": json.dumps({"error": "Invalid JSON body"}),
            }
    else:
        # Direct invocation
        body = event

    # Extract governance request
    action = body.get("action", "evaluate")
    payload = body.get("payload", {})
    agent_id = body.get("agent_id", "lambda-default")

    try:
        if action == "evaluate":
            # Govern the payload text. This runs entirely inside the Lambda.
            result = tork.govern(
                payload.get("text", ""),
                agent_id=agent_id,
                session_id=body.get("session_id"),
            )

            response_body = {
                # GovernanceResult fields: action, output, pii, receipt, report
                "action": result.action.value,
                "output": result.output,
                "pii_types": [t.value for t in result.pii.types],
                "pii_count": result.pii.count,
                "receipt_id": result.receipt.receipt_id,
                "request_id": context.aws_request_id,
            }

        elif action == "redact":
            # PII redaction only — no governance decision
            text = body.get("text", "")
            pii = detect_pii(text)

            response_body = {
                "redacted": redact_pii(text),
                "matches": [
                    {
                        "type": m.type.value,
                        "start": m.start_index,
                        "end": m.end_index,
                    }
                    for m in pii.matches
                ],
            }

        elif action == "health":
            response_body = {
                "status": "healthy",
                "function": context.function_name,
                "memory_mb": context.memory_limit_in_mb,
                "remaining_time_ms": context.get_remaining_time_in_millis(),
            }

        else:
            return {
                "statusCode": 400,
                "headers": {"Content-Type": "application/json"},
                "body": json.dumps({"error": f"Unknown action: {action}"}),
            }

        # Return response
        if "body" in event:
            # API Gateway format
            return {
                "statusCode": 200,
                "headers": {
                    "Content-Type": "application/json",
                    "X-Request-Id": context.aws_request_id,
                },
                "body": json.dumps(response_body),
            }
        else:
            # Direct invocation
            return response_body

    except Exception as e:
        error_response = {
            "error": str(e),
            "error_type": type(e).__name__,
            "request_id": context.aws_request_id,
        }

        if "body" in event:
            return {
                "statusCode": 500,
                "headers": {"Content-Type": "application/json"},
                "body": json.dumps(error_response),
            }
        else:
            raise

SAM Template

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Tork AI Governance Lambda Function

Globals:
  Function:
    Timeout: 30
    MemorySize: 512
    Runtime: python3.11
    Architectures:
      - arm64  # Graviton2 for better price/performance

Parameters:
  TorkApiKey:
    Type: String
    NoEcho: true
    Description: Tork API Key

Resources:
  TorkGovernanceFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: tork-governance
      CodeUri: src/
      Handler: lambda_handler.lambda_handler
      Description: AI governance evaluation function
      Environment:
        Variables:
          TORK_API_KEY: !Ref TorkApiKey
      Layers:
        - !Ref TorkLayer
      Events:
        ApiGateway:
          Type: Api
          Properties:
            Path: /governance/{proxy+}
            Method: ANY
            RestApiId: !Ref TorkApi
      Policies:
        - Version: '2012-10-17'
          Statement:
            - Effect: Allow
              Action:
                - logs:CreateLogGroup
                - logs:CreateLogStream
                - logs:PutLogEvents
              Resource: '*'

  TorkLayer:
    Type: AWS::Serverless::LayerVersion
    Properties:
      LayerName: tork-dependencies
      Description: Tork governance dependencies
      ContentUri: layer/
      CompatibleRuntimes:
        - python3.11
      CompatibleArchitectures:
        - arm64
    Metadata:
      BuildMethod: python3.11

  TorkApi:
    Type: AWS::Serverless::Api
    Properties:
      Name: tork-governance-api
      StageName: prod
      Cors:
        AllowOrigin: "'*'"
        AllowMethods: "'POST,GET,OPTIONS'"
        AllowHeaders: "'Content-Type,Authorization'"

Outputs:
  ApiEndpoint:
    Description: API Gateway endpoint URL
    Value: !Sub "https://${TorkApi}.execute-api.${AWS::Region}.amazonaws.com/prod/"
  FunctionArn:
    Description: Lambda function ARN
    Value: !GetAtt TorkGovernanceFunction.Arn

#!/bin/bash
# Build Tork Lambda Layer for ARM64 (Graviton2)

set -e

LAYER_DIR="layer"
PYTHON_VERSION="3.11"

echo "Building Tork Lambda Layer..."

# Clean previous build
rm -rf $LAYER_DIR
mkdir -p $LAYER_DIR/python

# Install dependencies into layer
pip install \
  --platform manylinux2014_aarch64 \
  --implementation cp \
  --python-version $PYTHON_VERSION \
  --only-binary=:all: \
  --target $LAYER_DIR/python \
  tork-governance

# Create layer zip
cd $LAYER_DIR
zip -r ../tork-layer.zip .
cd ..

echo "Layer built: tork-layer.zip"
echo "Size: $(du -h tork-layer.zip | cut -f1)"

# Publish layer
aws lambda publish-layer-version \
  --layer-name tork-governance \
  --description "Tork AI Governance Library" \
  --zip-file fileb://tork-layer.zip \
  --compatible-runtimes python3.11 \
  --compatible-architectures arm64

Optimized requirements.txt

# Minimal dependencies for Lambda
tork>=1.0.0
# Exclude optional heavy dependencies
# --no-deps for tork, then add only what's needed:
pydantic>=2.0.0
regex>=2023.0.0
# Don't include: torch, transformers (use API-based detection instead)

"""
Cold-start optimized Lambda handler.
"""

import os

# Module-level initialization (persists across warm starts)
_tork = None


def get_tork():
    """Lazy initialization of Tork."""
    global _tork
    if _tork is None:
        from tork_governance import Tork
        _tork = Tork(api_key=os.environ.get("TORK_API_KEY"))
    return _tork


def lambda_handler(event, context):
    """Handler with lazy loading."""
    action = event.get("action", "evaluate")

    if action == "evaluate":
        # govern() is synchronous and returns a GovernanceResult dataclass
        result = get_tork().govern(event.get("payload", {}).get("text", ""))
        return {
            "action": result.action.value,
            "output": result.output,
            "receipt_id": result.receipt.receipt_id,
        }

    elif action == "redact":
        from tork_governance import redact_pii
        return {"redacted": redact_pii(event.get("text", ""))}

    return {"error": "Unknown action"}

Enable Provisioned Concurrency

# In SAM template
TorkGovernanceFunction:
  Type: AWS::Serverless::Function
  Properties:
    # ... other properties
    ProvisionedConcurrencyConfig:
      ProvisionedConcurrentExecutions: 5
    AutoPublishAlias: live
    DeploymentPreference:
      Type: AllAtOnce

"""
Google Cloud Function for Tork AI Governance.
"""

import functions_framework
import json
import os
from flask import jsonify, Request

from tork_governance import Tork, redact_pii


# Initialize globally for instance reuse
tork = Tork(api_key=os.environ.get("TORK_API_KEY"))


@functions_framework.http
def governance_handler(request: Request):
    """
    HTTP Cloud Function for governance evaluation.

    Args:
        request: Flask request object

    Returns:
        JSON response with governance result
    """
    # Handle CORS
    if request.method == "OPTIONS":
        headers = {
            "Access-Control-Allow-Origin": "*",
            "Access-Control-Allow-Methods": "POST, GET, OPTIONS",
            "Access-Control-Allow-Headers": "Content-Type, Authorization",
            "Access-Control-Max-Age": "3600",
        }
        return ("", 204, headers)

    headers = {"Access-Control-Allow-Origin": "*"}

    try:
        request_json = request.get_json(silent=True)

        if not request_json:
            return jsonify({"error": "No JSON body provided"}), 400, headers

        action = request_json.get("action", "evaluate")
        payload = request_json.get("payload", {})

        if action == "evaluate":
            result = tork.govern(
                payload.get("text", ""),
                agent_id=request_json.get("agent_id", "gcf-default"),
            )

            return jsonify({
                "action": result.action.value,
                "output": result.output,
                "pii_types": [t.value for t in result.pii.types],
                "pii_count": result.pii.count,
                "receipt_id": result.receipt.receipt_id,
            }), 200, headers

        elif action == "redact":
            text = request_json.get("text", "")
            return jsonify({"redacted": redact_pii(text)}), 200, headers

        elif action == "health":
            return jsonify({"status": "healthy"}), 200, headers

        else:
            return jsonify({"error": f"Unknown action: {action}"}), 400, headers

    except Exception as e:
        return jsonify({
            "error": str(e),
            "error_type": type(e).__name__,
        }), 500, headers


@functions_framework.cloud_event
def pubsub_governance(cloud_event):
    """
    Pub/Sub triggered governance for async processing.

    Triggered by messages to a Pub/Sub topic for batch
    governance evaluation.
    """
    import base64

    # Decode Pub/Sub message
    message_data = base64.b64decode(cloud_event.data["message"]["data"]).decode()
    payload = json.loads(message_data)

    # Evaluate governance
    result = tork.govern(
        payload.get("text", ""),
        agent_id=payload.get("agent_id", "pubsub-default"),
    )

    # Log metadata only (or publish to another topic) — never the text itself
    print(json.dumps({
        "event_id": cloud_event["id"],
        "action": result.action.value,
        "pii_count": result.pii.count,
        "receipt_id": result.receipt.receipt_id,
    }))

"""
Azure Functions app for Tork AI Governance.
"""

import azure.functions as func
import json
import os
import logging

from tork_governance import Tork, redact_pii


app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)

# Initialize Tork
tork = Tork(api_key=os.environ.get("TORK_API_KEY"))


@app.route(route="governance", methods=["POST"])
def governance_http(req: func.HttpRequest) -> func.HttpResponse:
    """
    HTTP trigger for governance evaluation.
    """
    logging.info("Governance evaluation request received")

    try:
        req_body = req.get_json()
    except ValueError:
        return func.HttpResponse(
            json.dumps({"error": "Invalid JSON body"}),
            status_code=400,
            mimetype="application/json",
        )

    action = req_body.get("action", "evaluate")
    payload = req_body.get("payload", {})

    try:
        if action == "evaluate":
            result = tork.govern(
                payload.get("text", ""),
                agent_id=req_body.get("agent_id", "azure-default"),
            )

            return func.HttpResponse(
                json.dumps({
                    "action": result.action.value,
                    "output": result.output,
                    "pii_count": result.pii.count,
                    "receipt_id": result.receipt.receipt_id,
                }),
                status_code=200,
                mimetype="application/json",
            )

        elif action == "redact":
            text = req_body.get("text", "")
            return func.HttpResponse(
                json.dumps({"redacted": redact_pii(text)}),
                status_code=200,
                mimetype="application/json",
            )

        else:
            return func.HttpResponse(
                json.dumps({"error": f"Unknown action: {action}"}),
                status_code=400,
                mimetype="application/json",
            )

    except Exception as e:
        logging.error(f"Governance error: {e}")
        return func.HttpResponse(
            json.dumps({"error": str(e)}),
            status_code=500,
            mimetype="application/json",
        )


@app.blob_trigger(
    arg_name="blob",
    path="ai-requests/{name}",
    connection="AzureWebJobsStorage"
)
def governance_blob(blob: func.InputStream):
    """
    Blob trigger for batch governance processing.

    Processes AI request files uploaded to blob storage.
    """
    logging.info(f"Processing blob: {blob.name}")

    content = blob.read().decode("utf-8")
    requests = json.loads(content)

    results = []
    for req in requests:
        result = tork.govern(
            req.get("payload", {}).get("text", ""),
            agent_id=req.get("agent_id", "blob-batch"),
        )
        results.append({
            "id": req.get("id"),
            "action": result.action.value,
            "pii_count": result.pii.count,
        })

    logging.info(f"Processed {len(results)} requests")
    # Results can be written to another blob or queue


@app.queue_trigger(
    arg_name="msg",
    queue_name="governance-queue",
    connection="AzureWebJobsStorage"
)
def governance_queue(msg: func.QueueMessage):
    """
    Queue trigger for async governance processing.
    """
    payload = json.loads(msg.get_body().decode("utf-8"))

    result = tork.govern(
        payload.get("payload", {}).get("text", ""),
        agent_id=payload.get("agent_id", "queue-default"),
    )

    logging.info(f"Queue message processed: action={result.action.value}")

/**
 * Vercel Edge Function for Tork AI Governance
 *
 * Runs at the edge for low-latency governance evaluation.
 */

import { NextRequest, NextResponse } from 'next/server';

// Edge runtime configuration
export const runtime = 'edge';
export const preferredRegion = ['iad1', 'sfo1', 'fra1']; // Multi-region

// Tork governance client (edge-compatible)
const TORK_API_URL = process.env.TORK_API_URL || 'https://api.tork.ai';
const TORK_API_KEY = process.env.TORK_API_KEY;

interface GovernanceRequest {
  action: 'evaluate' | 'redact' | 'health';
  payload?: Record<string, unknown>;
  text?: string;
  agent_id?: string;
}

interface GovernanceResult {
  decision: 'allow' | 'block' | 'modify';
  violations: string[];
  modified_payload?: Record<string, unknown>;
  pii_found?: Array<{ type: string; redacted: boolean }>;
}

async function evaluateGovernance(
  payload: Record<string, unknown>,
  agentId: string
): Promise<GovernanceResult> {
  const response = await fetch(`${TORK_API_URL}/v1/evaluate`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${TORK_API_KEY}`,
    },
    body: JSON.stringify({
      payload,
      agent_id: agentId,
    }),
  });

  if (!response.ok) {
    throw new Error(`Tork API error: ${response.status}`);
  }

  return response.json();
}

async function redactPII(text: string): Promise<{ redacted: string; matches: unknown[] }> {
  const response = await fetch(`${TORK_API_URL}/v1/redact`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${TORK_API_KEY}`,
    },
    body: JSON.stringify({ text }),
  });

  if (!response.ok) {
    throw new Error(`Tork API error: ${response.status}`);
  }

  return response.json();
}

export async function POST(request: NextRequest) {
  const startTime = Date.now();

  try {
    const body: GovernanceRequest = await request.json();
    const { action = 'evaluate', payload, text, agent_id = 'vercel-edge' } = body;

    let result: unknown;

    switch (action) {
      case 'evaluate':
        if (!payload) {
          return NextResponse.json(
            { error: 'Missing payload for evaluate action' },
            { status: 400 }
          );
        }
        result = await evaluateGovernance(payload, agent_id);
        break;

      case 'redact':
        if (!text) {
          return NextResponse.json(
            { error: 'Missing text for redact action' },
            { status: 400 }
          );
        }
        result = await redactPII(text);
        break;

      case 'health':
        result = {
          status: 'healthy',
          region: process.env.VERCEL_REGION || 'unknown',
          latency_ms: Date.now() - startTime,
        };
        break;

      default:
        return NextResponse.json(
          { error: `Unknown action: ${action}` },
          { status: 400 }
        );
    }

    return NextResponse.json({
      ...result as object,
      _meta: {
        latency_ms: Date.now() - startTime,
        region: process.env.VERCEL_REGION,
      },
    });
  } catch (error) {
    console.error('Governance error:', error);
    return NextResponse.json(
      {
        error: error instanceof Error ? error.message : 'Unknown error',
        _meta: {
          latency_ms: Date.now() - startTime,
          region: process.env.VERCEL_REGION,
        },
      },
      { status: 500 }
    );
  }
}

export async function GET() {
  return NextResponse.json({
    status: 'healthy',
    service: 'tork-governance-edge',
    region: process.env.VERCEL_REGION,
  });
}

Middleware for Automatic Governance

/**
 * Next.js Middleware for automatic AI API governance.
 *
 * Intercepts requests to /api/ai/* and applies governance
 * before forwarding to the AI provider.
 */

import { NextRequest, NextResponse } from 'next/server';

export const config = {
  matcher: '/api/ai/:path*',
};

const TORK_API_URL = process.env.TORK_API_URL || 'https://api.tork.ai';
const TORK_API_KEY = process.env.TORK_API_KEY;

export async function middleware(request: NextRequest) {
  // Only process POST requests with JSON body
  if (request.method !== 'POST') {
    return NextResponse.next();
  }

  try {
    const body = await request.json();

    // Evaluate against governance policies
    const governanceResponse = await fetch(`${TORK_API_URL}/v1/evaluate`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${TORK_API_KEY}`,
      },
      body: JSON.stringify({
        payload: body,
        agent_id: 'vercel-middleware',
        context: {
          path: request.nextUrl.pathname,
          method: request.method,
        },
      }),
    });

    if (!governanceResponse.ok) {
      console.error('Governance API error:', governanceResponse.status);
      // YOUR choice: this example proceeds on a non-2xx. Return a 503 here
      // instead if an ungoverned request is not acceptable in your system.
      return NextResponse.next();
    }

    const result = await governanceResponse.json();

    // Block if governance denies
    if (result.decision === 'block') {
      return NextResponse.json(
        {
          error: 'Request blocked by governance policy',
          violations: result.violations,
        },
        { status: 403 }
      );
    }

    // Forward with modified payload if needed
    if (result.decision === 'modify' && result.modified_payload) {
      const modifiedRequest = new NextRequest(request.url, {
        method: request.method,
        headers: request.headers,
        body: JSON.stringify(result.modified_payload),
      });
      return NextResponse.next({ request: modifiedRequest });
    }

    return NextResponse.next();
  } catch (error) {
    console.error('Middleware error:', error);
    // This example fails open: on an error in YOUR middleware the request
    // proceeds ungoverned. That is an error-handling decision you make about
    // your own code, not a property of Tork. If an ungoverned request is
    // unacceptable, fail closed here instead:
    //   return NextResponse.json({ error: 'Governance unavailable' }, { status: 503 });
    return NextResponse.next();
  }
}

"""
S3-triggered Lambda for document governance.

Scans uploaded documents for sensitive content before
they're processed by AI systems.
"""

import json
import boto3
import urllib.parse

from tork_governance import Tork, GovernanceAction


s3 = boto3.client("s3")
# No api_key: the decision is made inside this Lambda and nothing leaves it.
tork = Tork()


def lambda_handler(event, context):
    """
    Process S3 upload events for governance.

    Triggered when files are uploaded to the AI input bucket.
    Scans content and moves to appropriate destination.
    """
    results = []

    for record in event["Records"]:
        bucket = record["s3"]["bucket"]["name"]
        key = urllib.parse.unquote_plus(record["s3"]["object"]["key"])

        print(f"Processing: s3://{bucket}/{key}")

        # Download file content
        response = s3.get_object(Bucket=bucket, Key=key)
        content = response["Body"].read()

        # Determine content type
        content_type = response.get("ContentType", "application/octet-stream")

        # Only text-like objects can be governed as text
        if content_type.startswith("text/") or key.endswith((".json", ".txt", ".md")):
            text_content = content.decode("utf-8")
            result = tork.govern(text_content, agent_id="s3-ingest")
        else:
            # Binary file — nothing to scan; route it without a text decision
            result = None

        # Determine action based on governance
        if result is None:
            destination_bucket = f"{bucket}-unscanned"
            action = "unscanned"
        elif result.action == GovernanceAction.DENY:
            # Move to quarantine bucket
            destination_bucket = f"{bucket}-quarantine"
            action = "quarantined"
        elif result.action == GovernanceAction.REDACT:
            # Save the redacted version instead of the original
            destination_bucket = f"{bucket}-processed"
            content = result.output.encode("utf-8")
            action = "redacted"
        elif result.action == GovernanceAction.ESCALATE:
            destination_bucket = f"{bucket}-review"
            action = "escalated"
        else:
            # Move to approved bucket
            destination_bucket = f"{bucket}-approved"
            action = "approved"

        # Copy to destination. Metadata is labels and counts only — never
        # the detected PII values themselves.
        metadata = {"governance-action": action}
        if result is not None:
            metadata["governance-receipt-id"] = result.receipt.receipt_id
            metadata["governance-pii-types"] = ",".join(t.value for t in result.pii.types)
            metadata["governance-pii-count"] = str(result.pii.count)

        s3.put_object(
            Bucket=destination_bucket,
            Key=key,
            Body=content,
            Metadata=metadata,
        )

        # Delete from source
        s3.delete_object(Bucket=bucket, Key=key)

        results.append({
            "key": key,
            "action": action,
            "pii_count": result.pii.count if result else 0,
        })

        print(f"Processed {key}: {action}")

    return {
        "processed": len(results),
        "results": results,
    }

API Gateway Integration

# SAM template for API Gateway with Lambda authorizer

Resources:
  GovernanceApi:
    Type: AWS::Serverless::Api
    Properties:
      StageName: prod
      Auth:
        DefaultAuthorizer: TorkAuthorizer
        Authorizers:
          TorkAuthorizer:
            FunctionArn: !GetAtt AuthorizerFunction.Arn

  AuthorizerFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: authorizer.lambda_handler
      Runtime: python3.11
      CodeUri: src/
      Environment:
        Variables:
          TORK_API_KEY: !Ref TorkApiKey

  # Authorizer that checks governance before allowing request
  # authorizer.py content:
  #
  # def lambda_handler(event, context):
  #     token = event.get("authorizationToken", "")
  #     method_arn = event["methodArn"]
  #
  #     # Validate token and check governance
  #     if is_valid_and_governed(token):
  #         return generate_policy("user", "Allow", method_arn)
  #     else:
  #         return generate_policy("user", "Deny", method_arn)

  AIProxyFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: proxy.lambda_handler
      Runtime: python3.11
      CodeUri: src/
      Timeout: 120
      MemorySize: 1024
      Events:
        ProxyApi:
          Type: Api
          Properties:
            RestApiId: !Ref GovernanceApi
            Path: /ai/{proxy+}
            Method: ANY

DynamoDB Streams for Audit Logging

"""
DynamoDB Streams handler for governance audit logging.

Captures all governance decisions and stores them for
compliance and analytics.
"""

import json
import boto3
from datetime import datetime

dynamodb = boto3.resource("dynamodb")
audit_table = dynamodb.Table("tork-governance-audit")


def lambda_handler(event, context):
    """
    Process DynamoDB stream events for audit logging.
    """
    for record in event["Records"]:
        if record["eventName"] not in ["INSERT", "MODIFY"]:
            continue

        # Extract governance decision from new image
        new_image = record["dynamodb"].get("NewImage", {})

        # Parse DynamoDB format to regular dict
        decision_data = parse_dynamodb_item(new_image)

        # Create audit record. These are the fields the on-device SDK
        # actually gives you: the action, the receipt id, and PII labels
        # and counts — never the PII values themselves.
        audit_record = {
            "audit_id": f"audit-{context.aws_request_id}-{record['eventID']}",
            "timestamp": datetime.utcnow().isoformat(),
            "event_type": record["eventName"],
            "action": decision_data.get("action"),
            "receipt_id": decision_data.get("receipt_id"),
            "agent_id": decision_data.get("agent_id"),
            "pii_types": decision_data.get("pii_types", []),
            "pii_detected": decision_data.get("pii_count", 0),
            "source_table": record["eventSourceARN"].split("/")[1],
        }

        # Store in your own audit table. This is your log, in your account —
        # it is separate from anything Tork records.
        audit_table.put_item(Item=audit_record)

    return {"processed": len(event["Records"])}


def parse_dynamodb_item(item):
    """Convert DynamoDB item format to regular dict."""
    result = {}
    for key, value in item.items():
        if "S" in value:
            result[key] = value["S"]
        elif "N" in value:
            result[key] = int(value["N"])
        elif "L" in value:
            result[key] = [parse_dynamodb_item(v) if "M" in v else v.get("S") for v in value["L"]]
        elif "M" in value:
            result[key] = parse_dynamodb_item(value["M"])
    return result

"""
Batch processor for cost-efficient governance.

Processes multiple requests in a single Lambda invocation
to reduce per-request overhead.
"""

import json
from concurrent.futures import ThreadPoolExecutor

from tork_governance import Tork


tork = Tork()


def lambda_handler(event, context):
    """
    Process batch of governance requests.

    Accepts array of requests and processes them
    in parallel for efficiency.
    """
    requests = event.get("requests", [])

    if not requests:
        return {"error": "No requests provided", "processed": 0}

    # govern() is synchronous CPU-bound work on this machine, so a thread
    # pool overlaps it with nothing else here — it is used to keep the shape
    # simple. Drop the pool if you are CPU-bound in a small Lambda.
    results = []
    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = [
            executor.submit(process_request, req, i)
            for i, req in enumerate(requests)
        ]

        for future in futures:
            results.append(future.result())

    # Aggregate statistics over the real action values
    allowed = sum(1 for r in results if r["action"] == "allow")
    denied = sum(1 for r in results if r["action"] == "deny")
    redacted = sum(1 for r in results if r["action"] == "redact")

    return {
        "processed": len(results),
        "summary": {
            "allowed": allowed,
            "denied": denied,
            "redacted": redacted,
        },
        "results": results,
    }


def process_request(request, index):
    """Process a single governance request."""
    try:
        result = tork.govern(
            request.get("payload", {}).get("text", ""),
            agent_id=request.get("agent_id", f"batch-{index}"),
        )
        return {
            "index": index,
            "action": result.action.value,
            "pii_count": result.pii.count,
            "receipt_id": result.receipt.receipt_id,
            "success": True,
        }
    except Exception as e:
        return {
            "index": index,
            "action": "error",
            "error": str(e),
            "success": False,
        }

"""
Lambda handler with CloudWatch metrics integration.
"""

import os
import time
import boto3

from tork_governance import Tork


cloudwatch = boto3.client("cloudwatch")
tork = Tork()

NAMESPACE = "Tork/Governance"


def put_metric(name, value, unit="Count", dimensions=None):
    """Publish metric to CloudWatch."""
    metric_data = {
        "MetricName": name,
        "Value": value,
        "Unit": unit,
    }
    if dimensions:
        metric_data["Dimensions"] = [
            {"Name": k, "Value": v} for k, v in dimensions.items()
        ]

    cloudwatch.put_metric_data(
        Namespace=NAMESPACE,
        MetricData=[metric_data],
    )


def lambda_handler(event, context):
    """Handler with metrics instrumentation."""
    start_time = time.time()

    try:
        payload = event.get("payload", {})
        agent_id = event.get("agent_id", "default")

        # Evaluate governance
        result = tork.govern(payload.get("text", ""), agent_id=agent_id)

        # Record metrics
        duration_ms = (time.time() - start_time) * 1000

        put_metric("Invocations", 1, dimensions={"AgentId": agent_id})
        put_metric("Duration", duration_ms, unit="Milliseconds")
        put_metric(f"Action_{result.action.value.title()}", 1, dimensions={"AgentId": agent_id})
        put_metric("PIIDetected", result.pii.count)

        if result.action.value == "deny":
            put_metric("DeniedRequests", 1, dimensions={"AgentId": agent_id})

        return {
            "action": result.action.value,
            "receipt_id": result.receipt.receipt_id,
            "duration_ms": duration_ms,
        }

    except Exception as e:
        put_metric("Errors", 1)
        raise

CloudWatch Dashboard

{
  "widgets": [
    {
      "type": "metric",
      "properties": {
        "title": "Governance Decisions",
        "metrics": [
          ["Tork/Governance", "Action_Allow", {"color": "#22c55e"}],
          [".", "Action_Deny", {"color": "#ef4444"}],
          [".", "Action_Redact", {"color": "#f59e0b"}]
        ],
        "period": 60,
        "stat": "Sum"
      }
    },
    {
      "type": "metric",
      "properties": {
        "title": "Invocation Latency",
        "metrics": [
          ["Tork/Governance", "Duration", {"stat": "p50"}],
          [".", ".", {"stat": "p99"}]
        ],
        "period": 60
      }
    },
    {
      "type": "metric",
      "properties": {
        "title": "Cold Starts",
        "metrics": [
          ["AWS/Lambda", "ConcurrentExecutions", "FunctionName", "tork-governance"],
          [".", "Invocations", ".", "."]
        ],
        "period": 60
      }
    },
    {
      "type": "metric",
      "properties": {
        "title": "Error Rate",
        "metrics": [
          ["Tork/Governance", "Errors"],
          ["AWS/Lambda", "Errors", "FunctionName", "tork-governance"]
        ],
        "period": 60
      }
    }
  ]
}