AEGIS Production Deployment Guide¶

pip install aegis-governance

pip install aegis-governance[engine,telemetry,config]

pip install aegis-governance[all]

from aegis_governance import AegisConfig

# Uses frozen defaults matching schema/interface-contract.yaml
config = AegisConfig.default()

# Requires: pip install aegis-governance[config]
config = AegisConfig.from_yaml("config.yaml")

parameters:
  epsilon_R: 0.01
  epsilon_P: 0.01
  risk_trigger_factor: 2.0
  profit_trigger_factor: 2.0
  trigger_confidence_prob: 0.95
  novelty_gate:
    N0: 0.7
    k: 10.0
    output_threshold: 0.8
  complexity_floor: 0.5
  quality_min_score: 0.7

config = AegisConfig.from_dict({"epsilon_R": 0.02, "quality_min_score": 0.8})

docker build -t aegis-governance .

docker compose up -d

apiVersion: apps/v1
kind: Deployment
metadata:
  name: aegis-governance
  labels:
    app: aegis-governance
spec:
  replicas: 2
  selector:
    matchLabels:
      app: aegis-governance
  template:
    metadata:
      labels:
        app: aegis-governance
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "9090"
        prometheus.io/path: "/metrics"
    spec:
      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
      containers:
        - name: aegis
          image: aegis-governance:latest
          command: ["python", "-c", "import signal; from telemetry.metrics_server import MetricsServer; s=MetricsServer(host='0.0.0.0',port=9090); s.start(); signal.sigwait({signal.SIGTERM,signal.SIGINT}); s.stop()"]
          ports:
            - containerPort: 9090
              name: metrics
          resources:
            requests:
              cpu: "100m"
              memory: "128Mi"
            limits:
              cpu: "500m"
              memory: "256Mi"
          livenessProbe:
            exec:
              command: ["aegis", "version"]
            initialDelaySeconds: 5
            periodSeconds: 30
          readinessProbe:
            httpGet:
              path: /metrics
              port: 9090
            initialDelaySeconds: 5
            periodSeconds: 10
          env:
            - name: AEGIS_CONFIG_PATH
              value: "/app/config/config.yaml"
            - name: DATABASE_URL
              valueFrom:
                secretKeyRef:
                  name: aegis-secrets
                  key: database-url
                  optional: true
          volumeMounts:
            - name: config
              mountPath: /app/config
              readOnly: true
      volumes:
        - name: config
          configMap:
            name: aegis-config

apiVersion: v1
kind: ConfigMap
metadata:
  name: aegis-config
data:
  config.yaml: |
    parameters:
      epsilon_R: 0.01
      epsilon_P: 0.01
      risk_trigger_factor: 2.0
      profit_trigger_factor: 2.0
      trigger_confidence_prob: 0.95

apiVersion: v1
kind: Secret
metadata:
  name: aegis-secrets
type: Opaque
stringData:
  database-url: "postgresql+asyncpg://{USER}:{PASSWORD}@{HOST}:5432/aegis"
  # HSM credentials (if PQ-hardened profile)
  hsm-pin: "{HSM_PIN}"

apiVersion: v1
kind: Service
metadata:
  name: aegis-governance
  labels:
    app: aegis-governance
spec:
  selector:
    app: aegis-governance
  ports:
    - name: metrics
      port: 9090
      targetPort: 9090
---
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: aegis-governance
  labels:
    release: prometheus
spec:
  selector:
    matchLabels:
      app: aegis-governance
  endpoints:
    - port: metrics
      interval: 30s
      path: /metrics

import json
from aegis_governance import AegisConfig, PCWContext, PCWPhase, pcw_decide

config = AegisConfig.default()
evaluator = config.create_gate_evaluator()

def handler(event, context):
    ctx = PCWContext(
        agent_id=event.get("agent_id", "lambda"),
        session_id=context.aws_request_id,
        phase=PCWPhase.PLAN,
        proposal_summary=event["proposal_summary"],
        estimated_impact=event.get("estimated_impact", "medium"),
        risk_proposed=event.get("risk_score", 0.1),
        complexity_score=event.get("complexity_score", 0.5),
        quality_score=event.get("quality_score", 0.8),
    )
    decision = pcw_decide(ctx, gate_evaluator=evaluator)
    return {
        "statusCode": 200,
        "body": json.dumps({
            "status": decision.status.value,
            "rationale": decision.rationale,
            "decision_id": decision.decision_id,
        }),
    }

{
  "family": "aegis-governance",
  "networkMode": "awsvpc",
  "requiresCompatibilities": ["FARGATE"],
  "cpu": "256",
  "memory": "512",
  "containerDefinitions": [
    {
      "name": "aegis",
      "image": "aegis-governance:latest",
      "command": ["python", "-c", "import signal; from telemetry.metrics_server import MetricsServer; s=MetricsServer(host='0.0.0.0',port=9090); s.start(); signal.sigwait({signal.SIGTERM,signal.SIGINT}); s.stop()"],
      "portMappings": [
        { "containerPort": 9090, "protocol": "tcp" }
      ],
      "environment": [
        { "name": "AEGIS_METRICS_PORT", "value": "9090" }
      ],
      "healthCheck": {
        "command": ["CMD-SHELL", "aegis version || exit 1"],
        "interval": 30,
        "timeout": 5,
        "retries": 3
      }
    }
  ]
}

# Programmatic (persistent server in background thread)
from telemetry.metrics_server import MetricsServer
server = MetricsServer(port=9090)
server.start()

aegis metrics

echo '{"risk_proposed": 0.3, "profit_proposed": 0.1}' | \
  aegis evaluate --telemetry-url https://yd1xm4ahcg.execute-api.us-west-2.amazonaws.com/dev/v1/events

telemetry_url: "https://yd1xm4ahcg.execute-api.us-west-2.amazonaws.com/dev/v1/events"

from aegis_governance import HTTPEventSink, TelemetryEmitter, pcw_decide

emitter = TelemetryEmitter(source="my-service")
emitter.add_sink(HTTPEventSink(
    url="https://yd1xm4ahcg.execute-api.us-west-2.amazonaws.com/dev/v1/events",
    headers={"Authorization": "Bearer ${TOKEN}"},
    timeout=10,
))
result = pcw_decide(context, telemetry_emitter=emitter)

from aegis_governance import http_sink

sink = http_sink(
    "https://yd1xm4ahcg.execute-api.us-west-2.amazonaws.com/dev/v1/events",
    batch_size=100,
    flush_interval_seconds=60,
)
sink.start()  # Background flush thread
emitter.add_sink(sink)
# ... on shutdown:
sink.stop()   # Drains remaining events

from telemetry.emitter import HTTPEventSink

# Explicitly opt into insecure transport for local development
sink = HTTPEventSink(
    url="http://localhost:9090/events",
    allow_insecure=True,  # Logs a WARNING for audit trail
)

# 1. Verify system health
aegis health

# 2. Resume pending workflows (programmatic)
from workflows.persistence.durable import DurableWorkflowEngine
engine = DurableWorkflowEngine(session_factory)
await engine.resume_all_pending()