┌─────────────────────────────────────────────────────────────┐
│                    Load Balancer                             │
│                  (Round Robin / Least Conn)                 │
└───────────────────────┬─────────────────────────────────────┘
                        │
        ┌───────────────┼───────────────┐
        │               │               │
┌───────▼──────┐ ┌──────▼──────┐ ┌──────▼──────┐
│ Agent        │ │ Agent       │ │ Agent       │
│ Instance 1   │ │ Instance 2  │ │ Instance N  │
│              │ │             │ │             │
│ ┌──────────┐ │ │ ┌──────────┐│ │ ┌──────────┐│
│ │ Intent   │ │ │ │ Intent   ││ │ │ Intent   ││
│ │ Handler  │ │ │ │ Handler  ││ │ │ Handler  ││
│ └────┬─────┘ │ │ └────┬─────┘│ │ └────┬─────┘│
│      │       │ │      │      │ │      │       │
│ ┌────▼─────┐ │ │ ┌────▼─────┐│ │ ┌────▼─────┐│
│ │ Context  │ │ │ │ Context  ││ │ │ Context  ││
│ │ Service  │ │ │ │ Service  ││ │ │ Service  ││
│ └────┬─────┘ │ │ └────┬─────┘│ │ └────┬─────┘│
│      │       │ │      │      │ │      │       │
│ ┌────▼─────┐ │ │ ┌────▼─────┐│ │ ┌────▼─────┐│
│ │ LLM      │ │ │ │ LLM      ││ │ │ LLM      ││
│ │ Client   │ │ │ │ Client   ││ │ │ Client   ││
│ └────┬─────┘ │ │ └────┬─────┘│ │ └────┬─────┘│
│      │       │ │      │      │ │      │       │
│ ┌────▼─────┐ │ │ ┌────▼─────┐│ │ ┌────▼─────┐│
│ │ Tool     │ │ │ │ Tool     ││ │ │ Tool     ││
│ │ Executor │ │ │ │ Executor ││ │ │ Executor ││
│ └────┬─────┘ │ │ └────┬─────┘│ │ └────┬─────┘│
│      │       │ │      │      │ │      │       │
│ ┌────▼─────┐ │ │ ┌────▼─────┐│ │ ┌────▼─────┐│
│ │ Response │ │ │ │ Response ││ │ │ Response ││
│ │ Builder  │ │ │ │ Builder  ││ │ │ Builder  ││
│ └──────────┘ │ │ └──────────┘│ │ └──────────┘│
└──────────────┘ └─────────────┘ └──────────────┘
        │               │               │
        └───────────────┼───────────────┘
                        │
┌───────────────────────▼─────────────────────────────────────┐
│              Shared Resources Layer                        │
│  ┌────────────┐ ┌────────────┐ ┌────────────┐            │
│  │ PostgreSQL │ │    Redis     │ │  Message   │            │
│  │ (State)    │ │   (Cache)    │ │   Queue    │            │
│  └────────────┘ └────────────┘ └────────────┘            │
└─────────────────────────────────────────────────────────────┘

# app/main.py
from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
import asyncio
import structlog
from typing import Optional, Dict, Any, List
from datetime import datetime

from .services.intent_classifier import IntentClassifier
from .services.context_retriever import ContextRetriever
from .services.llm_client import LLMClient, LLMResponse
from .services.tool_executor import ToolExecutor
from .services.response_builder import ResponseBuilder
from .models.state import ConversationState, StateManager
from .models.request import AgentRequest, AgentResponse
from .config import Settings, get_settings
from .middleware import RateLimitMiddleware, AuthenticationMiddleware
from .observability import setup_tracing, record_metrics

logger = structlog.get_logger()

# Global service instances (initialized once, shared across requests)
services: Dict[str, Any] = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Initialize and cleanup application resources."""
    settings = get_settings()
    
    # Initialize services
    logger.info("initializing_services")
    services["intent_classifier"] = IntentClassifier(settings)
    services["context_retriever"] = ContextRetriever(settings)
    services["llm_client"] = LLMClient(settings)
    services["tool_executor"] = ToolExecutor(settings)
    services["response_builder"] = ResponseBuilder(settings)
    services["state_manager"] = StateManager(settings)
    
    # Setup observability
    setup_tracing(settings)
    
    yield
    
    # Cleanup
    logger.info("shutting_down_services")
    for service in services.values():
        if hasattr(service, 'cleanup'):
            await service.cleanup()

app = FastAPI(
    title="AI Agent Service",
    description="Production-ready AI agent monolith",
    version="1.0.0",
    lifespan=lifespan
)

# Middleware
app.add_middleware(CORSMiddleware, allow_origins=["*"])
app.add_middleware(AuthenticationMiddleware)
app.add_middleware(RateLimitMiddleware, requests_per_minute=100)

@app.post("/v1/agent/process", response_model=AgentResponse)
async def process_request(
    request: AgentRequest,
    background_tasks: BackgroundTasks,
    settings: Settings = Depends(get_settings)
) -> AgentResponse:
    """Process a single agent request through the complete pipeline."""
    start_time = datetime.utcnow()
    request_id = request.request_id or generate_request_id()
    
    log = logger.bind(request_id=request_id, user_id=request.user_id)
    log.info("request_received", intent_hint=request.message[:50])
    
    try:
        # Stage 1: Intent Classification
        with record_metrics("intent_classification"):
            intent = await services["intent_classifier"].classify(
                message=request.message,
                user_context=request.context
            )
        log.info("intent_classified", intent_type=intent.type, confidence=intent.confidence)
        
        # Stage 2: Context Retrieval
        with record_metrics("context_retrieval"):
            context = await services["context_retriever"].retrieve(
                query=request.message,
                intent=intent,
                user_id=request.user_id,
                conversation_id=request.conversation_id
            )
        log.info("context_retrieved", docs_count=len(context.documents))
        
        # Stage 3: Load Conversation State
        state = await services["state_manager"].get_state(
            conversation_id=request.conversation_id
        )
        
        # Stage 4: LLM Processing with Tool Support
        with record_metrics("llm_processing"):
            llm_response = await services["llm_client"].generate(
                message=request.message,
                intent=intent,
                context=context,
                conversation_history=state.history,
                tools=services["tool_executor"].get_available_tools(intent),
                timeout_ms=settings.llm_timeout_ms,
                max_tokens=settings.max_response_tokens
            )
        log.info("llm_response_generated", tokens_used=llm_response.tokens_used)
        
        # Stage 5: Tool Execution (if requested by LLM)
        tool_results = []
        if llm_response.tool_calls:
            with record_metrics("tool_execution"):
                tool_results = await services["tool_executor"].execute_all(
                    tool_calls=llm_response.tool_calls,
                    user_context=request.context
                )
            log.info("tools_executed", tool_count=len(tool_results))
            
            # Re-run LLM with tool results
            llm_response = await services["llm_client"].generate_with_tools(
                original_message=request.message,
                tool_results=tool_results,
                context=context
            )
        
        # Stage 6: Response Building
        with record_metrics("response_building"):
            response = await services["response_builder"].build(
                llm_response=llm_response,
                intent=intent,
                tool_results=tool_results,
                user_preferences=state.preferences
            )
        
        # Stage 7: Update State (async, non-blocking)
        background_tasks.add_task(
            update_conversation_state,
            conversation_id=request.conversation_id,
            user_message=request.message,
            assistant_response=response.content,
            metadata={
                "intent": intent.type,
                "tokens_used": llm_response.tokens_used,
                "tools_used": [t.tool_name for t in tool_results]
            }
        )
        
        # Calculate total processing time
        processing_time_ms = (datetime.utcnow() - start_time).total_seconds() * 1000
        
        log.info("request_completed", 
                 processing_time_ms=processing_time_ms,
                 response_length=len(response.content))
        
        return AgentResponse(
            request_id=request_id,
            content=response.content,
            metadata=AgentMetadata(
                intent_type=intent.type,
                confidence=intent.confidence,
                processing_time_ms=processing_time_ms,
                tokens_used=llm_response.tokens_used,
                sources=[doc.source for doc in context.documents]
            )
        )
        
    except asyncio.TimeoutError:
        log.error("request_timeout", timeout_ms=settings.llm_timeout_ms)
        raise HTTPException(status_code=504, detail="Request timeout")
    except Exception as e:
        log.error("request_failed", error=str(e), exc_info=True)
        raise HTTPException(status_code=500, detail="Internal processing error")

async def update_conversation_state(
    conversation_id: str,
    user_message: str,
    assistant_response: str,
    metadata: Dict[str, Any]
):
    """Update conversation state asynchronously."""
    try:
        await services["state_manager"].append_turn(
            conversation_id=conversation_id,
            user_message=user_message,
            assistant_response=assistant_response,
            metadata=metadata
        )
    except Exception as e:
        logger.error("state_update_failed", 
                    conversation_id=conversation_id, 
                    error=str(e))

@app.get("/health")
async def health_check():
    """Health check endpoint for load balancers."""
    return {"status": "healthy", "services": list(services.keys())}

@app.get("/ready")
async def readiness_check():
    """Readiness check for Kubernetes."""
    # Verify all services are initialized
    for name, service in services.items():
        if hasattr(service, 'is_healthy') and not await service.is_healthy():
            raise HTTPException(status_code=503, detail=f"Service {name} unhealthy")
    return {"status": "ready"}

# Dockerfile
FROM python:3.12-slim-bookworm

# Security: Run as non-root user
RUN groupadd -r appgroup && useradd -r -g appgroup appuser

# Install system dependencies
RUN apt-get update && apt-get install -y \
    gcc \
    libpq-dev \
    && rm -rf /var/lib/apt/lists/*

# Set working directory
WORKDIR /app

# Copy and install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY app/ ./app/

# Change ownership to non-root user
RUN chown -R appuser:appgroup /app
USER appuser

# Expose port
EXPOSE 8000

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD python -c "import requests; requests.get('http://localhost:8000/health')"

# Run with uvicorn
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

# docker-compose.yml (for local development)
version: '3.8'
services:
  agent-service:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@postgres:5432/agentdb
      - REDIS_URL=redis://redis:6379
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - LOG_LEVEL=INFO
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy
    deploy:
      replicas: 2
      resources:
        limits:
          cpus: '2'
          memory: 2G
        reservations:
          cpus: '1'
          memory: 1G

  postgres:
    image: postgres:16-alpine
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
      - POSTGRES_DB=agentdb
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user -d agentdb"]
      interval: 5s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 5s
      retries: 5

volumes:
  postgres_data:
  redis_data:

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-service
  labels:
    app: agent-service
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  selector:
    matchLabels:
      app: agent-service
  template:
    metadata:
      labels:
        app: agent-service
    spec:
      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
        fsGroup: 1000
      containers:
      - name: agent-service
        image: tropicalmedia/agent-service:v1.0.0
        imagePullPolicy: Always
        ports:
        - containerPort: 8000
          name: http
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: agent-secrets
              key: database-url
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: agent-secrets
              key: openai-api-key
        - name: REDIS_URL
          value: "redis://redis-cluster:6379"
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "2000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5
        volumeMounts:
        - name: tmp
          mountPath: /tmp
      volumes:
      - name: tmp
        emptyDir: {}
---
apiVersion: v1
kind: Service
metadata:
  name: agent-service
spec:
  selector:
    app: agent-service
  ports:
  - port: 80
    targetPort: 8000
  type: ClusterIP
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: agent-service-ingress
  annotations:
    kubernetes.io/ingress.class: nginx
    nginx.ingress.kubernetes.io/rate-limit: "100"
    cert-manager.io/cluster-issuer: letsencrypt
spec:
  tls:
  - hosts:
    - agents.tropical-media.work
    secretName: agent-service-tls
  rules:
  - host: agents.tropical-media.work
    http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: agent-service
            port:
              number: 80

┌──────────────────────────────────────────────────────────────┐
│                        API Gateway                            │
│           (Auth, Rate Limiting, Request Routing)             │
└───────────────────────────┬──────────────────────────────────┘
                            │
        ┌───────────────────┼───────────────────┐
        │                   │                   │
┌───────▼──────┐   ┌───────▼──────┐   ┌───────▼──────┐
│   Intent     │   │   Context    │   │  Response    │
│  Service     │   │   Service    │   │   Service    │
│  (3 replicas)│   │  (10 replicas)│   │  (3 replicas)│
└───────┬──────┘   └───────┬──────┘   └───────┬──────┘
        │                   │                   │
        └───────────────────┼───────────────────┘
                            │
                    ┌───────▼──────┐
                    │   Reasoning  │
                    │   Service    │
                    │ (5 replicas) │
                    └───────┬──────┘
                            │
        ┌───────────────────┼───────────────────┐
        │                   │                   │
┌───────▼──────┐   ┌───────▼──────┐   ┌───────▼──────┐
│    Tool      │   │    State     │   │   Metrics    │
│   Service    │   │   Service    │   │   Service    │
│  (2 replicas)│   │  (3 replicas)│   │  (2 replicas)│
└──────────────┘   └──────────────┘   └──────────────┘

# intent-service/main.py
from fastapi import FastAPI
from pydantic import BaseModel
import joblib
import numpy as np

app = FastAPI()

# Load lightweight classification model (DistilBERT or similar)
model = joblib.load("models/intent_classifier.pkl")

class IntentRequest(BaseModel):
    message: str
    user_id: str
    session_context: dict

class IntentResponse(BaseModel):
    intent: str
    confidence: float
    entities: list
    requires_tool: bool

@app.post("/classify", response_model=IntentResponse)
async def classify_intent(request: IntentRequest):
    # Fast, lightweight classification (sub-50ms)
    prediction = model.predict([request.message])[0]
    confidence = model.predict_proba([request.message])[0].max()
    
    return IntentResponse(
        intent=prediction.intent_type,
        confidence=confidence,
        entities=extract_entities(request.message),
        requires_tool=prediction.requires_tool
    )

# context-service/main.py
import asyncpg
from qdrant_client import AsyncQdrantClient
import redis.asyncio as redis

class ContextService:
    def __init__(self):
        self.qdrant = AsyncQdrantClient(host="qdrant", port=6333)
        self.pg = asyncpg.create_pool(dsn=os.environ["DATABASE_URL"])
        self.cache = redis.Redis(host="redis", port=6379)
    
    async def retrieve_context(self, query: str, intent: str, user_id: str):
        # Check cache first
        cache_key = f"ctx:{user_id}:{hash(query)}"
        cached = await self.cache.get(cache_key)
        if cached:
            return json.loads(cached)
        
        # Query vector store for semantic matches
        vector_results = await self.qdrant.search(
            collection_name="knowledge_base",
            query_vector=await self.embed(query),
            limit=5
        )
        
        # Query PostgreSQL for structured data
        async with self.pg.acquire() as conn:
            user_data = await conn.fetch(
                "SELECT * FROM user_profiles WHERE user_id = $1",
                user_id
            )
        
        context = ContextResult(
            documents=vector_results,
            user_profile=user_data,
            retrieved_at=datetime.utcnow()
        )
        
        # Cache for 5 minutes
        await self.cache.setex(
            cache_key, 300, json.dumps(context.dict())
        )
        
        return context

# reasoning-service/main.py
import openai
from langchain.chains import ConversationChain
from langchain.memory import ConversationBufferMemory

class ReasoningService:
    def __init__(self):
        self.client = openai.AsyncOpenAI()
        self.memories = {}  # In production, use Redis or external store
    
    async def reason(
        self, 
        message: str, 
        context: ContextResult,
        conversation_id: str,
        available_tools: list
    ):
        # Retrieve or create memory
        if conversation_id not in self.memories:
            self.memories[conversation_id] = ConversationBufferMemory()
        
        memory = self.memories[conversation_id]
        
        # Build enhanced prompt with context
        prompt = build_reasoning_prompt(
            message=message,
            context=context,
            memory=memory,
            tools=available_tools
        )
        
        # Call LLM with retry logic
        for attempt in range(3):
            try:
                response = await self.client.chat.completions.create(
                    model="gpt-4o",
                    messages=[{"role": "user", "content": prompt}],
                    tools=available_tools,
                    timeout=30
                )
                break
            except openai.Timeout:
                if attempt == 2:
                    raise
                await asyncio.sleep(2 ** attempt)
        
        # Update memory
        memory.save_context(
            {"input": message},
            {"output": response.choices[0].message.content}
        )
        
        return ReasoningResult(
            content=response.choices[0].message.content,
            tool_calls=response.choices[0].message.tool_calls,
            tokens_used=response.usage.total_tokens
        )

# Using httpx for async HTTP calls between services
import httpx

async def call_context_service(query: str, user_id: str):
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "http://context-service:8000/retrieve",
            json={"query": query, "user_id": user_id},
            timeout=5.0
        )
        return response.json()

# Using RabbitMQ for async processing
import aio_pika

async def publish_document_processing_job(document_id: str):
    connection = await aio_pika.connect_robust("amqp://rabbitmq")
    async with connection:
        channel = await connection.channel()
        exchange = await channel.declare_exchange(
            "agent_jobs", aio_pika.ExchangeType.TOPIC
        )
        
        message = aio_pika.Message(
            json.dumps({
                "document_id": document_id,
                "operation": "index",
                "priority": "normal"
            }).encode(),
            delivery_mode=aio_pika.DeliveryMode.PERSISTENT
        )
        
        await exchange.publish(message, routing_key="document.processing")

# istio/virtual-service.yaml
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: agent-services
spec:
  hosts:
  - "*.agent-cluster.local"
  http:
  - match:
    - uri:
        prefix: /intent
    route:
    - destination:
        host: intent-service
        port:
          number: 8000
      weight: 100
    timeout: 2s
    retries:
      attempts: 3
      perTryTimeout: 1s
      retryOn: gateway-error,connect-failure,refused-stream
  - match:
    - uri:
        prefix: /context
    route:
    - destination:
        host: context-service
        port:
          number: 8000
      weight: 100
    fault:
      delay:
        percentage:
          value: 0.1
        fixedDelay: 5s

# lambda_function.py
import json
import boto3
import os
from aws_lambda_powertools import Logger, Tracer, Metrics
from aws_lambda_powertools.metrics import MetricUnit
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.event_handler import APIGatewayRestResolver

logger = Logger()
tracer = Tracer()
metrics = Metrics()
app = APIGatewayRestResolver()

# Initialize services outside handler for connection reuse
s3 = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')
state_table = dynamodb.Table(os.environ['STATE_TABLE_NAME'])

@logger.inject_lambda_context(correlation_id_path=correlation_paths.API_GATEWAY_REST)
@tracer.capture_lambda_handler
@metrics.log_metrics(capture_cold_start_metric=True)
def lambda_handler(event, context):
    """Main Lambda entry point."""
    return app.resolve(event, context)

@app.post("/agent/process")
@tracer.capture_method
async def process_agent_request():
    """Process an agent request."""
    request_id = app.context.get("aws_request_id")
    body = app.current_event.json_body
    
    metrics.add_dimension(name="agent_type", value=body.get("agent_type", "default"))
    
    try:
        # Load conversation state from DynamoDB
        state = await load_state(body.get("conversation_id"))
        
        # Process through agent pipeline
        with tracer.provider.in_subsegment('## process_pipeline'):
            result = await process_pipeline(
                message=body["message"],
                state=state,
                user_id=body["user_id"]
            )
        
        # Save updated state
        await save_state(body["conversation_id"], result.state)
        
        metrics.add_metric(name="SuccessCount", unit=MetricUnit.Count, value=1)
        metrics.add_metric(name="ProcessingLatency", 
                          unit=MetricUnit.Milliseconds, 
                          value=result.processing_time_ms)
        
        return {
            "statusCode": 200,
            "body": json.dumps({
                "request_id": request_id,
                "response": result.content,
                "tokens_used": result.tokens_used
            })
        }
        
    except Exception as e:
        logger.exception("Processing failed")
        metrics.add_metric(name="ErrorCount", unit=MetricUnit.Count, value=1)
        return {
            "statusCode": 500,
            "body": json.dumps({"error": "Processing failed"})
        }

@tracer.capture_method
async def process_pipeline(message: str, state: dict, user_id: str):
    """Core agent processing pipeline."""
    # Import here to minimize cold start impact
    from agent.intent import classify_intent
    from agent.context import retrieve_context
    from agent.llm import generate_response
    
    # Pipeline stages
    intent = await classify_intent(message, state)
    context = await retrieve_context(intent, user_id)
    response = await generate_response(message, intent, context, state)
    
    return response

@tracer.capture_method
async def load_state(conversation_id: str):
    """Load conversation state from DynamoDB."""
    if not conversation_id:
        return {}
    
    response = state_table.get_item(Key={'conversation_id': conversation_id})
    return response.get('Item', {}).get('state', {})

@tracer.capture_method  
async def save_state(conversation_id: str, state: dict):
    """Save conversation state to DynamoDB."""
    state_table.put_item(
        Item={
            'conversation_id': conversation_id,
            'state': state,
            'updated_at': datetime.utcnow().isoformat(),
            'ttl': int((datetime.utcnow() + timedelta(days=30)).timestamp())
        }
    )

# serverless.yml
service: agent-serverless

provider:
  name: aws
  runtime: python3.12
  region: eu-central-1
  memorySize: 2048  # Higher memory for faster LLM processing
  timeout: 30
  environment:
    OPENAI_API_KEY: ${ssm:/agent/openai-api-key}
    STATE_TABLE_NAME: ${self:service}-state-${sls:stage}
    LOG_LEVEL: INFO
  iam:
    role:
      statements:
        - Effect: Allow
          Action:
            - dynamodb:GetItem
            - dynamodb:PutItem
            - dynamodb:UpdateItem
          Resource:
            - !GetStateTableArn
        - Effect: Allow
          Action:
            - s3:GetObject
            - s3:PutObject
          Resource:
            - !GetDocumentsBucketArn

plugins:
  - serverless-python-requirements

custom:
  pythonRequirements:
    dockerizePip: true
    slim: true
    strip: false
    layer: true

functions:
  agentApi:
    handler: lambda_function.lambda_handler
    events:
      - http:
          path: /agent/{proxy+}
          method: ANY
          cors: true
          authorizer:
            name: cognitoAuthorizer
            arn: ${cf:auth-stack.CognitoUserPoolArn}
    layers:
      - {Ref: PythonRequirementsLambdaLayer}
  
  documentProcessor:
    handler: handlers/document_processor.handler
    memorySize: 4096  # Higher for document processing
    timeout: 300  # 5 minutes for large documents
    events:
      - s3:
          bucket: ${self:service}-documents-${sls:stage}
          event: s3:ObjectCreated:*
          rules:
            - prefix: uploads/
            - suffix: .pdf
  
  scheduledTasks:
    handler: handlers/scheduled.handler
    events:
      - schedule: rate(5 minutes)

resources:
  Resources:
    StateTable:
      Type: AWS::DynamoDB::Table
      Properties:
        TableName: ${self:service}-state-${sls:stage}
        BillingMode: PAY_PER_REQUEST
        AttributeDefinitions:
          - AttributeName: conversation_id
            AttributeType: S
        KeySchema:
          - AttributeName: conversation_id
            KeyType: HASH
        TimeToLiveSpecification:
          AttributeName: ttl
          Enabled: true
    
    DocumentsBucket:
      Type: AWS::S3::Bucket
      Properties:
        BucketName: ${self:service}-documents-${sls:stage}
        LifecycleConfiguration:
          Rules:
            - Id: ExpireOldUploads
              Status: Enabled
              ExpirationInDays: 30

functions:
  agentApi:
    handler: lambda_function.lambda_handler
    provisionedConcurrency: 10  # Always keep 10 instances warm

# Heavy imports deferred until needed
_llm_client = None

async def get_llm_client():
    global _llm_client
    if _llm_client is None:
        import openai
        _llm_client = openai.AsyncOpenAI()
    return _llm_client

// worker.ts
import { OpenAI } from 'openai';

export interface Env {
  OPENAI_API_KEY: string;
  VECTORIZE_INDEX: VectorizeIndex;
  KV_NAMESPACE: KVNamespace;
}

export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
    // Handle CORS
    if (request.method === 'OPTIONS') {
      return new Response(null, {
        headers: {
          'Access-Control-Allow-Origin': '*',
          'Access-Control-Allow-Methods': 'POST',
          'Access-Control-Allow-Headers': 'Content-Type, Authorization',
        },
      });
    }

    if (request.method !== 'POST') {
      return new Response('Method not allowed', { status: 405 });
    }

    try {
      const { message, conversation_id, user_id } = await request.json();
      
      // Load conversation state from KV (sub-millisecond latency)
      const state = await env.KV_NAMESPACE.get(`state:${conversation_id}`);
      const conversationHistory = state ? JSON.parse(state) : [];
      
      // Query vector index for context (runs at edge)
      const vectorResults = await env.VECTORIZE_INDEX.query(
        await generateEmbedding(message, env.OPENAI_API_KEY),
        { topK: 5 }
      );
      
      // Build context from vector results
      const context = vectorResults.matches
        .map(match => match.metadata?.text)
        .join('\n\n');
      
      // Call OpenAI API (closest edge location)
      const openai = new OpenAI({ apiKey: env.OPENAI_API_KEY });
      
      const completion = await openai.chat.completions.create({
        model: 'gpt-4o-mini',  // Use lighter model for edge
        messages: [
          {
            role: 'system',
            content: `You are a helpful assistant. Use this context to answer:\n\n${context}`
          },
          ...conversationHistory.slice(-5),  // Last 5 messages
          { role: 'user', content: message }
        ],
        max_tokens: 500,
      });
      
      const assistantMessage = completion.choices[0].message.content;
      
      // Update conversation state
      conversationHistory.push(
        { role: 'user', content: message },
        { role: 'assistant', content: assistantMessage }
      );
      
      // Save state asynchronously (don't block response)
      ctx.waitUntil(
        env.KV_NAMESPACE.put(
          `state:${conversation_id}`,
          JSON.stringify(conversationHistory),
          { expirationTtl: 86400 }  // 24 hour TTL
        )
      );
      
      return new Response(
        JSON.stringify({
          response: assistantMessage,
          conversation_id,
          sources: vectorResults.matches.map(m => m.id)
        }),
        {
          headers: {
            'Content-Type': 'application/json',
            'Access-Control-Allow-Origin': '*',
            'CF-Cache-Status': 'DYNAMIC'
          }
        }
      );
      
    } catch (error) {
      return new Response(
        JSON.stringify({ error: error.message }),
        { status: 500, headers: { 'Content-Type': 'application/json' } }
      );
    }
  }
};

async function generateEmbedding(text: string, apiKey: string): Promise<number[]> {
  const response = await fetch('https://api.openai.com/v1/embeddings', {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${apiKey}`,
      'Content-Type': 'application/json'
    },
    body: JSON.stringify({
      model: 'text-embedding-3-small',
      input: text
    })
  });
  
  const data = await response.json();
  return data.data[0].embedding;
}

# wrangler.toml
name = "agent-edge"
main = "worker.ts"
compatibility_date = "2024-06-01"

# Environment variables
[vars]
ENVIRONMENT = "production"

# Secrets (use `wrangler secret put` for these)
# OPENAI_API_KEY

[[vectorize]]
binding = "VECTORIZE_INDEX"
index_name = "knowledge-base"

[[kv_namespaces]]
binding = "KV_NAMESPACE"
id = "your-kv-namespace-id"

# Durable Objects for stateful sessions
[[durable_objects.bindings]]
name = "AGENT_SESSION"
class_name = "AgentSession"

[[migrations]]
tag = "v1"
new_classes = ["AgentSession"]

┌─────────────────────────────────────────────────────────────┐
│                      User Request                            │
│                    (Tokyo, Japan)                           │
└───────────────────────┬─────────────────────────────────────┘
                        │
                        ▼
┌─────────────────────────────────────────────────────────────┐
│              Cloudflare Edge Network                        │
│  ┌─────────────────────────────────────────────────────┐    │
│  │           Tokyo Data Center (250 locations)        │    │
│  │  ┌─────────────┐    ┌─────────────┐               │    │
│  │  │   Worker    │───▶│   Vector    │               │    │
│  │  │   Runtime   │    │   Index     │               │    │
│  │  └─────────────┘    └─────────────┘               │    │
│  │          │                  │                      │    │
│  │          ▼                  ▼                      │    │
│  │  ┌─────────────┐    ┌─────────────┐               │    │
│  │  │     KV      │    │  Durable    │               │    │
│  │  │   Store     │    │  Object     │               │    │
│  │  └─────────────┘    └─────────────┘               │    │
│  └─────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────┘
                        │
            ┌───────────┴───────────┐
            │                       │
            ▼                       ▼
┌──────────────────┐    ┌──────────────────┐
│  Origin Server   │    │  AI API (Origin) │
│  (if needed)     │    │  (OpenAI, etc.)  │
└──────────────────┘    └──────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                      CDN / Edge Layer                        │
│         (Cloudflare/AWS CloudFront - Static Assets)         │
└─────────────────────────────────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────────┐
│                   API Gateway / Load Balancer                │
└───────────────────────┬─────────────────────────────────────┘
                        │
        ┌───────────────┼───────────────┐
        │               │               │
        ▼               ▼               ▼
┌──────────┐    ┌────────────┐    ┌────────────┐
│  Edge    │    │ Kubernetes │    │ Serverless │
│ Workers  │    │   Cluster  │    │ Functions  │
│(Chat,   │    │(Core Agent │    │(Background │
│Queries)  │    │ Services)  │    │ Tasks)     │
└──────────┘    └────────────┘    └────────────┘
        │               │               │
        └───────────────┼───────────────┘
                        │
                ┌───────▼───────┐
                │ Data Layer    │
                │ (PostgreSQL,  │
                │ Redis, Qdrant)│
                └───────────────┘

# router.py
class HybridRouter:
    def __init__(self):
        self.edge_client = CloudflareWorkerClient()
        self.k8s_client = KubernetesServiceClient()
        self.lambda_client = LambdaClient()
    
    async def route_request(self, request: AgentRequest) -> RouteTarget:
        """Determine the best deployment target for this request."""
        
        # Simple queries with cached context -> Edge
        if self.is_simple_query(request) and self.has_cached_context(request):
            return RouteTarget.EDGE
        
        # Complex reasoning requiring tool execution -> Kubernetes
        if request.requires_tools or request.expected_complexity == "high":
            return RouteTarget.KUBERNETES
        
        # Document processing, async tasks -> Serverless
        if request.operation_type in ["document_ingestion", "batch_process"]:
            return RouteTarget.SERVERLESS
        
        # Default to Kubernetes for reliability
        return RouteTarget.KUBERNETES
    
    async def execute(self, request: AgentRequest):
        target = await self.route_request(request)
        
        routing_map = {
            RouteTarget.EDGE: self.edge_client.process,
            RouteTarget.KUBERNETES: self.k8s_client.process,
            RouteTarget.SERVERLESS: self.lambda_client.process
        }
        
        handler = routing_map.get(target)
        if not handler:
            raise RoutingError(f"Unknown route target: {target}")
        
        return await handler(request)

# failover-config.yaml
routing_rules:
  primary: kubernetes-cluster
  fallback_chain:
    - kubernetes-cluster
    - serverless-region-1
    - serverless-region-2
    - edge-workers
  
  health_checks:
    interval: 10s
    timeout: 5s
    unhealthy_threshold: 2
    healthy_threshold: 2
  
  failover_triggers:
    - error_rate > 5%
    - latency_p99 > 2000ms
    - availability < 99.9%

# k8s/complete-agent-stack.yaml

# Namespace for isolation
apiVersion: v1
kind: Namespace
metadata:
  name: agent-system
  labels:
    istio-injection: enabled

---
# ConfigMap for non-sensitive configuration
apiVersion: v1
kind: ConfigMap
metadata:
  name: agent-config
  namespace: agent-system
data:
  LOG_LEVEL: "INFO"
  MAX_CONCURRENT_REQUESTS: "100"
  LLM_TIMEOUT_MS: "30000"
  CACHE_TTL_SECONDS: "300"

---
# Secret for sensitive data
apiVersion: v1
kind: Secret
metadata:
  name: agent-secrets
  namespace: agent-system
type: Opaque
stringData:
  openai-api-key: "sk-..."
  database-password: "..."
  jwt-secret: "..."

---
# Persistent Volume for state storage
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: agent-state-pvc
  namespace: agent-system
spec:
  accessModes:
    - ReadWriteOnce
  resources:
    requests:
      storage: 50Gi
  storageClassName: fast-ssd

---
# Deployment with HPA
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-core
  namespace: agent-system
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 25%
      maxUnavailable: 10%
  selector:
    matchLabels:
      app: agent-core
  template:
    metadata:
      labels:
        app: agent-core
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "8000"
    spec:
      serviceAccountName: agent-sa
      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
      containers:
      - name: agent
        image: tropicalmedia/agent-core:v1.2.3
        imagePullPolicy: IfNotPresent
        ports:
        - containerPort: 8000
          name: http
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: agent-secrets
              key: database-url
        - name: LOG_LEVEL
          valueFrom:
            configMapKeyRef:
              name: agent-config
              key: LOG_LEVEL
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
            nvidia.com/gpu: "0"  # No GPU for this service
          limits:
            memory: "4Gi"
            cpu: "2000m"
        volumeMounts:
        - name: state-storage
          mountPath: /data/state
        - name: tmp
          mountPath: /tmp
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 60
          periodSeconds: 10
          failureThreshold: 3
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 5
        startupProbe:
          httpGet:
            path: /health
            port: 8000
          failureThreshold: 30
          periodSeconds: 10
      volumes:
      - name: state-storage
        persistentVolumeClaim:
          claimName: agent-state-pvc
      - name: tmp
        emptyDir: {}

---
# Horizontal Pod Autoscaler
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: agent-core-hpa
  namespace: agent-system
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: agent-core
  minReplicas: 3
  maxReplicas: 50
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
  - type: Pods
    pods:
      metric:
        name: agent_requests_per_second
      target:
        type: AverageValue
        averageValue: "100"
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
      - type: Percent
        value: 100
        periodSeconds: 15
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
      - type: Percent
        value: 10
        periodSeconds: 60

---
# Vertical Pod Autoscaler for right-sizing
apiVersion: autoscaling.k8s.io/v1
kind: VerticalPodAutoscaler
metadata:
  name: agent-core-vpa
  namespace: agent-system
spec:
  targetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: agent-core
  updatePolicy:
    updateMode: "Auto"
  resourcePolicy:
    containerPolicies:
    - containerName: agent
      minAllowed:
        cpu: 100m
        memory: 256Mi
      maxAllowed:
        cpu: 4
        memory: 8Gi
      controlledResources: ["cpu", "memory"]

---
# Pod Disruption Budget for availability
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
  name: agent-core-pdb
  namespace: agent-system
spec:
  minAvailable: 2
  selector:
    matchLabels:
      app: agent-core

---
# Service for internal communication
apiVersion: v1
kind: Service
metadata:
  name: agent-core
  namespace: agent-system
  labels:
    app: agent-core
spec:
  selector:
    app: agent-core
  ports:
  - port: 80
    targetPort: 8000
    name: http
  type: ClusterIP

---
# Network Policy for security
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: agent-network-policy
  namespace: agent-system
spec:
  podSelector:
    matchLabels:
      app: agent-core
  policyTypes:
  - Ingress
  - Egress
  ingress:
  - from:
    - namespaceSelector:
        matchLabels:
          name: ingress-nginx
    - podSelector:
        matchLabels:
          app: api-gateway
    ports:
    - protocol: TCP
      port: 8000
  egress:
  - to:
    - namespaceSelector:
        matchLabels:
          name: database
    ports:
    - protocol: TCP
      port: 5432
  - to:
    - namespaceSelector:
        matchLabels:
          name: redis
    ports:
    - protocol: TCP
      port: 6379
  - to:
    - namespaceSelector: {}  # Allow internet for LLM APIs
    ports:
    - protocol: TCP
      port: 443

# Chart.yaml
apiVersion: v2
name: agent-system
description: Production AI Agent System
type: application
version: 1.0.0
appVersion: "1.2.3"
dependencies:
  - name: postgresql
    version: "12.x.x"
    repository: "https://charts.bitnami.com/bitnami"
    condition: postgresql.enabled
  - name: redis
    version: "18.x.x"
    repository: "https://charts.bitnami.com/bitnami"
    condition: redis.enabled
  - name: qdrant
    version: "0.x.x"
    repository: "https://qdrant.github.io/qdrant-helm"
    condition: qdrant.enabled

# values.yaml
replicaCount: 3

image:
  repository: tropicalmedia/agent-core
  tag: v1.2.3
  pullPolicy: IfNotPresent

service:
  type: ClusterIP
  port: 80
  targetPort: 8000

resources:
  requests:
    cpu: 500m
    memory: 1Gi
  limits:
    cpu: 2000m
    memory: 4Gi

autoscaling:
  enabled: true
  minReplicas: 3
  maxReplicas: 50
  targetCPUUtilizationPercentage: 70
  targetMemoryUtilizationPercentage: 80

persistence:
  enabled: true
  size: 50Gi
  storageClass: fast-ssd

ingress:
  enabled: true
  className: nginx
  annotations:
    nginx.ingress.kubernetes.io/rate-limit: "100"
    cert-manager.io/cluster-issuer: letsencrypt
  hosts:
    - host: agents.tropical-media.work
      paths:
        - path: /
          pathType: Prefix
  tls:
    - secretName: agent-tls
      hosts:
        - agents.tropical-media.work

postgresql:
  enabled: true
  auth:
    existingSecret: agent-secrets
  primary:
    persistence:
      size: 100Gi
    resources:
      requests:
        cpu: 500m
        memory: 1Gi

redis:
  enabled: true
  auth:
    existingSecret: agent-secrets
    existingSecretPasswordKey: redis-password
  master:
    persistence:
      size: 20Gi

qdrant:
  enabled: true
  persistence:
    size: 200Gi
  resources:
    requests:
      cpu: 1000m
      memory: 4Gi

# istio/canary-deployment.yaml
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: agent-routing
  namespace: agent-system
spec:
  hosts:
  - agent-core
  http:
  - match:
    - headers:
        x-canary:
          exact: "true"
    route:
    - destination:
        host: agent-core
        subset: v2
      weight: 100
  - route:
    - destination:
        host: agent-core
        subset: v1
      weight: 90
    - destination:
        host: agent-core
        subset: v2
      weight: 10
    timeout: 5s
    retries:
      attempts: 3
      perTryTimeout: 2s
      retryOn: gateway-error,connect-failure,refused-stream
    fault:
      delay:
        percentage:
          value: 0.1
        fixedDelay: 100ms
---
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: agent-versions
  namespace: agent-system
spec:
  host: agent-core
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        http1MaxPendingRequests: 50
        maxRequestsPerConnection: 10
    loadBalancer:
      simple: LEAST_CONN
    outlierDetection:
      consecutiveErrors: 5
      interval: 30s
      baseEjectionTime: 30s
  subsets:
  - name: v1
    labels:
      version: v1.2.2
  - name: v2
    labels:
      version: v1.2.3

apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: llm-client-circuit-breaker
spec:
  host: openai-api.external
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        http1MaxPendingRequests: 50
        maxRequestsPerConnection: 2
    circuitBreaker:
      consecutiveErrors: 3
      interval: 10s
      baseEjectionTime: 30s
      maxEjectionPercent: 50

apiVersion: networking.istio.io/v1alpha3
kind: EnvoyFilter
metadata:
  name: agent-rate-limit
spec:
  configPatches:
  - applyTo: HTTP_FILTER
    match:
      context: SIDECAR_INBOUND
      listener:
        filterChain:
          filter:
            name: envoy.filters.network.http_connection_manager
    patch:
      operation: INSERT_BEFORE
      value:
        name: envoy.filters.http.local_ratelimit
        typed_config:
          "@type": type.googleapis.com/udpa.type.v1.TypedStruct
          type_url: type.googleapis.com/envoy.extensions.filters.http.local_ratelimit.v3.LocalRateLimit
          value:
            stat_prefix: http_local_rate_limiter
            token_bucket:
              max_tokens: 100
              tokens_per_fill: 10
              fill_interval: 1s
            filter_enabled:
              runtime_key: local_rate_limit_enabled
              default_value:
                numerator: 100
                denominator: HUNDRED
            filter_enforced:
              runtime_key: local_rate_limit_enforced
              default_value:
                numerator: 100
                denominator: HUNDRED

# event-driven-agent.py
import asyncio
import aio_pika
import json
from typing import Callable

class EventDrivenAgent:
    def __init__(self, amqp_url: str):
        self.amqp_url = amqp_url
        self.connection = None
        self.channel = None
        self.handlers: dict[str, Callable] = {}
    
    async def connect(self):
        self.connection = await aio_pika.connect_robust(self.amqp_url)
        self.channel = await self.connection.channel()
        await self.channel.set_qos(prefetch_count=10)
    
    def register_handler(self, event_type: str, handler: Callable):
        self.handlers[event_type] = handler
    
    async def start_consuming(self):
        """Start consuming events from multiple queues."""
        
        # Main request queue
        request_queue = await self.channel.declare_queue(
            "agent.requests",
            durable=True,
            arguments={"x-max-priority": 10}
        )
        
        # Priority queue for urgent requests
        priority_queue = await self.channel.declare_queue(
            "agent.requests.priority",
            durable=True,
            arguments={"x-max-priority": 10, "x-message-ttl": 300000}
        )
        
        # Dead letter queue for failed processing
        dlq = await self.channel.declare_queue(
            "agent.requests.dlq",
            durable=True
        )
        
        await request_queue.consume(self._process_message)
        await priority_queue.consume(self._process_priority_message)
        
        logger.info("Event-driven agent started consuming")
        
        # Keep running
        while True:
            await asyncio.sleep(1)
    
    async def _process_message(self, message: aio_pika.IncomingMessage):
        """Process a standard priority message."""
        async with message.process():
            try:
                event = json.loads(message.body)
                event_type = event.get("type")
                
                handler = self.handlers.get(event_type)
                if handler:
                    result = await handler(event["data"])
                    await self._publish_result(event["correlation_id"], result)
                else:
                    logger.warning(f"No handler for event type: {event_type}")
                    
            except Exception as e:
                logger.exception("Message processing failed")
                # Reject and requeue if retry count permits
                if message.headers.get("x-retry-count", 0) < 3:
                    await message.reject(requeue=False)
                    await self._republish_with_retry(event)
                else:
                    await message.reject(requeue=False)
    
    async def _process_priority_message(self, message: aio_pika.IncomingMessage):
        """Process high-priority messages with dedicated resources."""
        # Similar to _process_message but with priority handling
        pass
    
    async def _republish_with_retry(self, event: dict):
        """Republish message to retry queue with incremented counter."""
        retry_count = event.get("headers", {}).get("x-retry-count", 0) + 1
        event["headers"] = event.get("headers", {})
        event["headers"]["x-retry-count"] = retry_count
        
        # Exponential backoff
        delay_ms = min(1000 * (2 ** retry_count), 300000)  # Max 5 min
        
        await self.publish_event(
            "agent.requests.retry",
            event,
            headers={"x-delay": delay_ms}
        )
    
    async def publish_event(
        self, 
        routing_key: str, 
        data: dict, 
        headers: dict = None
    ):
        """Publish an event to the message bus."""
        message = aio_pika.Message(
            json.dumps(data).encode(),
            content_type="application/json",
            headers=headers,
            delivery_mode=aio_pika.DeliveryMode.PERSISTENT
        )
        
        await self.channel.default_exchange.publish(
            message,
            routing_key=routing_key
        )

# saga-orchestrator.py
from dataclasses import dataclass
from typing import List, Dict, Any, Callable
import asyncio

@dataclass
class SagaStep:
    name: str
    action: Callable
    compensation: Callable
    max_retries: int = 3

class SagaOrchestrator:
    def __init__(self):
        self.steps: List[SagaStep] = []
        self.completed_steps: List[str] = []
        self.failed_step: str = None
    
    def add_step(self, step: SagaStep):
        self.steps.append(step)
    
    async def execute(self, context: Dict[str, Any]) -> SagaResult:
        """Execute saga with compensation on failure."""
        for step in self.steps:
            try:
                # Execute step with retries
                for attempt in range(step.max_retries):
                    try:
                        result = await step.action(context)
                        context[f"{step.name}_result"] = result
                        self.completed_steps.append(step.name)
                        break
                    except RetryableError:
                        if attempt == step.max_retries - 1:
                            raise
                        await asyncio.sleep(2 ** attempt)
                        
            except Exception as e:
                self.failed_step = step.name
                logger.error(f"Saga step {step.name} failed: {e}")
                
                # Execute compensations in reverse order
                await self._compensate(context)
                
                return SagaResult(
                    success=False,
                    failed_step=step.name,
                    completed_steps=self.completed_steps
                )
        
        return SagaResult(
            success=True,
            completed_steps=self.completed_steps
        )
    
    async def _compensate(self, context: Dict[str, Any]):
        """Execute compensation actions in reverse order."""
        for step_name in reversed(self.completed_steps):
            step = next(s for s in self.steps if s.name == step_name)
            try:
                await step.compensation(context)
                logger.info(f"Compensation for {step_name} executed successfully")
            except Exception as e:
                logger.error(f"Compensation for {step_name} failed: {e}")
                # Alert for manual intervention
                await self._alert_manual_intervention(step_name, e)

# Example usage
def create_order_workflow():
    saga = SagaOrchestrator()
    
    saga.add_step(SagaStep(
        name="validate_customer",
        action=validate_customer_exists,
        compensation=noop_compensation
    ))
    
    saga.add_step(SagaStep(
        name="check_inventory",
        action=check_inventory_available,
        compensation=release_inventory_hold
    ))
    
    saga.add_step(SagaStep(
        name="process_payment",
        action=charge_customer,
        compensation=refund_payment
    ))
    
    saga.add_step(SagaStep(
        name="create_shipment",
        action=schedule_shipment,
        compensation=cancel_shipment
    ))
    
    return saga

# state-management.py
from abc import ABC, abstractmethod
from typing import Optional, Dict, Any
import json
import pickle
from datetime import datetime, timedelta

class StateBackend(ABC):
    """Abstract base class for state persistence backends."""
    
    @abstractmethod
    async def get(self, key: str) -> Optional[Dict[str, Any]]:
        pass
    
    @abstractmethod
    async def set(self, key: str, value: Dict[str, Any], ttl: Optional[int] = None):
        pass
    
    @abstractmethod
    async def delete(self, key: str):
        pass

class RedisStateBackend(StateBackend):
    """Redis-based state storage with serialization."""
    
    def __init__(self, redis_client):
        self.redis = redis_client
        self.serialization_format = "json"  # or "pickle" for complex objects
    
    async def get(self, key: str) -> Optional[Dict[str, Any]]:
        data = await self.redis.get(f"state:{key}")
        if data is None:
            return None
        
        if self.serialization_format == "json":
            return json.loads(data)
        else:
            return pickle.loads(data)
    
    async def set(self, key: str, value: Dict[str, Any], ttl: Optional[int] = None):
        if self.serialization_format == "json":
            data = json.dumps(value)
        else:
            data = pickle.dumps(value)
        
        if ttl:
            await self.redis.setex(f"state:{key}", ttl, data)
        else:
            await self.redis.set(f"state:{key}", data)
    
    async def delete(self, key: str):
        await self.redis.delete(f"state:{key}")

class PostgreSQLStateBackend(StateBackend):
    """PostgreSQL-based state for durability and complex queries."""
    
    def __init__(self, pool):
        self.pool = pool
    
    async def get(self, key: str) -> Optional[Dict[str, Any]]:
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow(
                """
                SELECT state_data, updated_at 
                FROM agent_states 
                WHERE state_key = $1 
                AND (expires_at IS NULL OR expires_at > NOW())
                """,
                key
            )
            return json.loads(row["state_data"]) if row else None
    
    async def set(self, key: str, value: Dict[str, Any], ttl: Optional[int] = None):
        expires_at = None
        if ttl:
            expires_at = datetime.utcnow() + timedelta(seconds=ttl)
        
        async with self.pool.acquire() as conn:
            await conn.execute(
                """
                INSERT INTO agent_states (state_key, state_data, expires_at, updated_at)
                VALUES ($1, $2, $3, NOW())
                ON CONFLICT (state_key) 
                DO UPDATE SET 
                    state_data = EXCLUDED.state_data,
                    expires_at = EXCLUDED.expires_at,
                    updated_at = NOW()
                """,
                key, json.dumps(value), expires_at
            )

class TieredStateManager:
    """Multi-tier state management: L1 (Memory) -> L2 (Redis) -> L3 (PostgreSQL)."""
    
    def __init__(self, memory_cache, redis_backend, pg_backend):
        self.l1 = memory_cache  # asyncio.LRUCache or similar
        self.l2 = redis_backend
        self.l3 = pg_backend
    
    async def get(self, key: str) -> Optional[Dict[str, Any]]:
        # Try L1 (memory) first
        value = self.l1.get(key)
        if value is not None:
            return value
        
        # Try L2 (Redis)
        value = await self.l2.get(key)
        if value is not None:
            # Populate L1
            self.l1.put(key, value)
            return value
        
        # Try L3 (PostgreSQL)
        value = await self.l3.get(key)
        if value is not None:
            # Populate L2 and L1
            await self.l2.set(key, value, ttl=3600)  # 1 hour in Redis
            self.l1.put(key, value)
            return value
        
        return None
    
    async def set(self, key: str, value: Dict[str, Any], ttl: Optional[int] = None):
        # Write-through to all layers
        self.l1.put(key, value, ttl=ttl)
        await self.l2.set(key, value, ttl=ttl)
        await self.l3.set(key, value, ttl=ttl)

# load-balancer.py
import asyncio
from typing import List, Dict
import random

class IntelligentLoadBalancer:
    """Application-aware load balancer for agent requests."""
    
    def __init__(self, backends: List[Backend]):
        self.backends = backends
        self.health_status = {b.id: True for b in backends}
        self.metrics = {b.id: BackendMetrics() for b in backends}
    
    async def select_backend(self, request: AgentRequest) -> Backend:
        """Select optimal backend based on request characteristics."""
        
        # Filter healthy backends
        healthy = [b for b in self.backends if self.health_status[b.id]]
        if not healthy:
            raise NoHealthyBackendsError()
        
        # Strategy based on request type
        if request.expected_duration == "long":
            # Least connections for long-running requests
            return min(healthy, key=lambda b: self.metrics[b.id].active_requests)
        
        elif request.priority == "high":
            # Lowest latency for high priority
            return min(healthy, key=lambda b: self.metrics[b.id].avg_response_time)
        
        elif request.requires_gpu:
            # GPU-capable backends only
            gpu_backends = [b for b in healthy if b.has_gpu]
            if gpu_backends:
                return random.choice(gpu_backends)
        
        # Default: weighted round-robin
        return self._weighted_round_robin(healthy)
    
    async def update_health(self):
        """Continuous health checking."""
        while True:
            for backend in self.backends:
                try:
                    healthy = await self._check_health(backend)
                    self.health_status[backend.id] = healthy
                except Exception:
                    self.health_status[backend.id] = False
            await asyncio.sleep(10)
    
    async def _check_health(self, backend: Backend) -> bool:
        """Perform health check against backend."""
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(
                    f"{backend.url}/health",
                    timeout=aiohttp.ClientTimeout(total=5)
                ) as response:
                    return response.status == 200
        except:
            return False

# queue-based-scaler.py
import asyncio
from kubernetes import client, config

class QueueBasedAutoscaler:
    """Custom autoscaler based on queue depth and processing rate."""
    
    def __init__(self, queue_name: str, deployment_name: str, namespace: str):
        self.queue_name = queue_name
        self.deployment_name = deployment_name
        self.namespace = namespace
        config.load_incluster_config()
        self.apps_api = client.AppsV1Api()
    
    async def scale_loop(self):
        """Main scaling loop."""
        while True:
            try:
                # Get queue metrics
                queue_depth = await self.get_queue_depth()
                processing_rate = await self.get_processing_rate()
                
                # Get current replicas
                deployment = self.apps_api.read_namespaced_deployment(
                    self.deployment_name, self.namespace
                )
                current_replicas = deployment.spec.replicas
                
                # Calculate desired replicas
                desired = self.calculate_desired_replicas(
                    queue_depth, processing_rate, current_replicas
                )
                
                # Apply scaling
                if desired != current_replicas:
                    await self.apply_scaling(desired)
                
            except Exception as e:
                logger.error(f"Scaling error: {e}")
            
            await asyncio.sleep(15)  # Check every 15 seconds
    
    def calculate_desired_replicas(
        self, 
        queue_depth: int, 
        processing_rate: float,
        current: int
    ) -> int:
        """Calculate desired replica count based on metrics."""
        
        if processing_rate == 0:
            # Avoid division by zero
            return max(current, 1)
        
        # Target: process queue in 60 seconds
        target_processing_time = 60
        required_rate = queue_depth / target_processing_time
        
        # Calculate replicas needed
        replicas_needed = required_rate / processing_rate * current
        
        # Apply bounds
        min_replicas = 2
        max_replicas = 100
        
        desired = int(round(replicas_needed))
        return max(min(desired, max_replicas), min_replicas)

# security-layers.yaml
security_layers:
  edge:
    - waf: "Cloudflare/AWS WAF"
    - ddos_protection: enabled
    - bot_management: enabled
    - tls: "1.3 only"
  
  network:
    - zero_trust: true
    - mTLS: required
    - network_policies: strict
    - segmentation: by_service
  
  application:
    - authentication: JWT/OAuth2
    - authorization: RBAC
    - input_validation: strict
    - output_sanitization: enabled
  
  data:
    - encryption_at_rest: AES-256
    - encryption_in_transit: TLS1.3
    - pii_masking: automatic
    - audit_logging: comprehensive

# secrets-management.py
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient
import boto3
from google.cloud import secretmanager

class SecretManager:
    """Unified secret management across cloud providers."""
    
    def __init__(self, provider: str, config: dict):
        self.provider = provider
        if provider == "azure":
            credential = DefaultAzureCredential()
            self.client = SecretClient(
                vault_url=config["vault_url"], 
                credential=credential
            )
        elif provider == "aws":
            self.client = boto3.client("secretsmanager")
        elif provider == "gcp":
            self.client = secretmanager.SecretManagerServiceClient()
    
    async def get_secret(self, name: str) -> str:
        """Retrieve secret by name."""
        if self.provider == "azure":
            secret = self.client.get_secret(name)
            return secret.value
        elif self.provider == "aws":
            response = self.client.get_secret_value(SecretId=name)
            return response["SecretString"]
        elif self.provider == "gcp":
            name = f"projects/{self.project}/secrets/{name}/versions/latest"
            response = self.client.access_secret_version(request={"name": name})
            return response.payload.data.decode("UTF-8")

# observability-setup.py
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from prometheus_client import Counter, Histogram, Gauge
import structlog

# Metrics
counter_requests = Counter(
    'agent_requests_total', 
    'Total agent requests', 
    ['agent_type', 'status']
)
histogram_latency = Histogram(
    'agent_request_duration_seconds',
    'Request latency',
    ['agent_type'],
    buckets=[.005, .01, .025, .05, .075, .1, .25, .5, .75, 1.0, 2.5, 5.0, 7.5, 10.0]
)
gauge_active_sessions = Gauge(
    'agent_active_sessions',
    'Number of active sessions'
)

# Tracing
tracer_provider = TracerProvider()
otlp_exporter = OTLPSpanExporter(endpoint="otel-collector:4317")
span_processor = BatchSpanProcessor(otlp_exporter)
tracer_provider.add_span_processor(span_processor)
trace.set_tracer_provider(tracer_provider)

# Structured Logging
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
        structlog.processors.JSONRenderer()
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

class ObservableAgent:
    """Agent with comprehensive observability."""
    
    def __init__(self):
        self.tracer = trace.get_tracer(__name__)
        self.logger = structlog.get_logger()
    
    async def process(self, request: AgentRequest):
        with self.tracer.start_as_current_span("agent.process") as span:
            span.set_attribute("agent.type", request.agent_type)
            span.set_attribute("user.id", request.user_id)
            
            start_time = time.time()
            
            try:
                # Processing logic
                result = await self._do_processing(request)
                
                counter_requests.labels(
                    agent_type=request.agent_type,
                    status="success"
                ).inc()
                
                span.set_attribute("result.status", "success")
                span.set_attribute("tokens.used", result.tokens_used)
                
            except Exception as e:
                counter_requests.labels(
                    agent_type=request.agent_type,
                    status="error"
                ).inc()
                span.set_attribute("error", True)
                span.set_attribute("error.message", str(e))
                raise
            
            finally:
                latency = time.time() - start_time
                histogram_latency.labels(agent_type=request.agent_type).observe(latency)
                span.set_attribute("duration_ms", latency * 1000)

# llm-observability.py
class LLMObservability:
    """Specialized observability for LLM operations."""
    
    def __init__(self):
        self.token_usage = Counter(
            'llm_tokens_total',
            'Total tokens used',
            ['model', 'token_type']  # prompt, completion
        )
        self.cost_usd = Counter(
            'llm_cost_usd_total',
            'Total LLM cost in USD',
            ['model']
        )
        self.latency = Histogram(
            'llm_request_duration_seconds',
            'LLM API latency',
            ['model']
        )
    
    def record_llm_call(
        self, 
        model: str, 
        prompt_tokens: int, 
        completion_tokens: int,
        latency_ms: float
    ):
        """Record LLM call metrics."""
        self.token_usage.labels(model=model, token_type="prompt").inc(prompt_tokens)
        self.token_usage.labels(model=model, token_type="completion").inc(completion_tokens)
        
        # Calculate cost (simplified pricing)
        cost_per_1k = self.get_pricing(model)
        total_tokens = prompt_tokens + completion_tokens
        cost = (total_tokens / 1000) * cost_per_1k
        self.cost_usd.labels(model=model).inc(cost)
        
        self.latency.labels(model=model).observe(latency_ms / 1000)

# multi-region-deployment.yaml
regions:
  primary:
    name: eu-central-1
    deployment: active
    rto_minutes: 5
    rpo_minutes: 1
    
  secondary:
    name: eu-west-1
    deployment: warm_standby
    rto_minutes: 15
    rpo_minutes: 5
    
  tertiary:
    name: us-east-1
    deployment: cold_standby
    rto_minutes: 60
    rpo_minutes: 15

data_replication:
  postgres:
    mode: streaming_replication
    lag_threshold_ms: 1000
    
  redis:
    mode: cross_region_replication
    
  object_storage:
    mode: geo_redundant
    
failover:
  automatic: true
  health_check_interval: 10s
  failure_threshold: 3
  dns_ttl: 60

# backup-manager.py
class AgentBackupManager:
    """Comprehensive backup for agent state and configuration."""
    
    async def create_backup(self):
        """Create point-in-time backup of all agent state."""
        timestamp = datetime.utcnow().isoformat()
        
        # Backup conversation states
        conversations = await self.export_conversations()
        await self.upload_to_s3(
            f"backups/{timestamp}/conversations.json",
            conversations
        )
        
        # Backup vector store
        vectors = await self.export_vector_store()
        await self.upload_to_s3(
            f"backups/{timestamp}/vectors.snapshot",
            vectors
        )
        
        # Backup configuration
        config = await self.export_configuration()
        await self.upload_to_s3(
            f"backups/{timestamp}/config.yaml",
            config
        )
        
        # Update backup manifest
        await self.update_manifest(timestamp)
    
    async def restore_from_backup(self, timestamp: str):
        """Restore system to specified backup point."""
        # Verify backup integrity
        if not await self.verify_backup(timestamp):
            raise BackupCorruptedError()
        
        # Restore in dependency order
        await self.restore_configuration(timestamp)
        await self.restore_vector_store(timestamp)
        await self.restore_conversations(timestamp)

# intelligent-caching.py
class LLMResponseCache:
    """Semantic caching for LLM responses."""
    
    def __init__(self, redis_client, embedding_model):
        self.redis = redis_client
        self.embedding_model = embedding_model
        self.similarity_threshold = 0.95
    
    async def get_cached_response(self, query: str) -> Optional[str]:
        """Check for semantically similar cached responses."""
        query_embedding = await self.embedding_model.embed(query)
        
        # Search vector cache
        results = await self.redis.ft.search(
            "llm_cache_index",
            query=f"*=>[KNN 5 @embedding $embedding]",
            query_params={"embedding": query_embedding}
        )
        
        for doc in results.docs:
            similarity = cosine_similarity(query_embedding, doc.embedding)
            if similarity > self.similarity_threshold:
                logger.info(f"Cache hit with similarity {similarity}")
                return doc.response
        
        return None
    
    async def cache_response(self, query: str, response: str, ttl: int = 86400):
        """Cache response with embedding."""
        embedding = await self.embedding_model.embed(query)
        
        await self.redis.hset(f"llm_cache:{hash(query)}", mapping={
            "query": query,
            "response": response,
            "embedding": embedding.tobytes(),
            "cached_at": datetime.utcnow().isoformat()
        })
        await self.redis.expire(f"llm_cache:{hash(query)}", ttl)

# model-tiering.py
class TieredModelRouter:
    """Route requests to appropriate model based on complexity."""
    
    def __init__(self):
        self.tiers = {
            "simple": {
                "model": "gpt-3.5-turbo",
                "cost_per_1k": 0.002,
                "max_tokens": 500
            },
            "standard": {
                "model": "gpt-4o-mini",
                "cost_per_1k": 0.015,
                "max_tokens": 2000
            },
            "complex": {
                "model": "gpt-4o",
                "cost_per_1k": 0.03,
                "max_tokens": 4000
            }
        }
    
    async def route_request(self, request: AgentRequest):
        """Select model tier based on request analysis."""
        
        # Analyze complexity
        complexity_score = await self.analyze_complexity(request.message)
        
        if complexity_score < 0.3 and len(request.message) < 200:
            tier = "simple"
        elif complexity_score < 0.7:
            tier = "standard"
        else:
            tier = "complex"
        
        config = self.tiers[tier]
        
        logger.info(f"Routing to {tier} tier: {config['model']}")
        
        return await self.call_model(config["model"], request)
    
    async def analyze_complexity(self, message: str) -> float:
        """Analyze message complexity (0-1 scale)."""
        # Simple heuristics - could be ML-based
        factors = [
            len(message) / 1000,  # Length factor
            message.count("?") / 3,  # Question complexity
            len([w for w in message.split() if len(w) > 8]) / 10,  # Complex words
        ]
        return min(sum(factors) / len(factors), 1.0)

# docker-compose.n8n-prod.yml
version: '3.8'
services:
  n8n:
    image: n8nio/n8n:latest
    restart: always
    environment:
      - N8N_BASIC_AUTH_ACTIVE=true
      - N8N_BASIC_AUTH_USER=${N8N_USER}
      - N8N_BASIC_AUTH_PASSWORD=${N8N_PASSWORD}
      - N8N_PORT=5678
      - N8N_PROTOCOL=https
      - NODE_ENV=production
      - WEBHOOK_URL=https://n8n.tropical-media.work/
      - GENERIC_TIMEZONE=Europe/Berlin
      - DB_TYPE=postgresdb
      - DB_POSTGRESDB_HOST=postgres
      - DB_POSTGRESDB_DATABASE=n8n
      - DB_POSTGRESDB_USER=n8n
      - DB_POSTGRESDB_PASSWORD=${POSTGRES_PASSWORD}
      - EXECUTIONS_MODE=queue
      - EXECUTIONS_TIMEOUT=3600
      - EXECUTIONS_DATA_MAX_AGE=336
      - QUEUE_BULL_REDIS_HOST=redis
      - QUEUE_BULL_REDIS_PORT=6379
      - QUEUE_HEALTH_CHECK_ACTIVE=true
      - N8N_METRICS=true
      - N8N_METRICS_PREFIX=n8n_
    volumes:
      - n8n_data:/home/node/.n8n
      - /var/run/docker.sock:/var/run/docker.sock
    deploy:
      replicas: 2
      resources:
        limits:
          cpus: '2'
          memory: 4G
    healthcheck:
      test: ["CMD", "wget", "--spider", "-q", "http://localhost:5678/healthz"]
      interval: 30s
      timeout: 10s
      retries: 3

  n8n-worker:
    image: n8nio/n8n:latest
    restart: always
    command: worker
    environment:
      - DB_TYPE=postgresdb
      - DB_POSTGRESDB_HOST=postgres
      - DB_POSTGRESDB_DATABASE=n8n
      - DB_POSTGRESDB_USER=n8n
      - DB_POSTGRESDB_PASSWORD=${POSTGRES_PASSWORD}
      - QUEUE_BULL_REDIS_HOST=redis
      - QUEUE_BULL_REDIS_PORT=6379
      - EXECUTIONS_TIMEOUT=3600
    deploy:
      replicas: 4
      resources:
        limits:
          cpus: '1'
          memory: 2G

  postgres:
    image: postgres:16-alpine
    restart: always
    environment:
      - POSTGRES_USER=n8n
      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
      - POSTGRES_DB=n8n
    volumes:
      - postgres_data:/var/lib/postgresql/data
    command: >
      postgres
      -c max_connections=200
      -c shared_buffers=2GB
      -c effective_cache_size=6GB

  redis:
    image: redis:7-alpine
    restart: always
    command: redis-server --appendonly yes --maxmemory 2gb --maxmemory-policy allkeys-lru
    volumes:
      - redis_data:/data

volumes:
  n8n_data:
  postgres_data:
  redis_data:

# openclaw-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: openclaw-gateway
  namespace: openclaw
spec:
  replicas: 2
  selector:
    matchLabels:
      app: openclaw-gateway
  template:
    metadata:
      labels:
        app: openclaw-gateway
    spec:
      containers:
      - name: gateway
        image: openclaw/gateway:v1.0.0
        ports:
        - containerPort: 3000
        env:
        - name: GATEWAY_BIND
          value: "0.0.0.0:3000"
        - name: GATEWAY_DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: openclaw-secrets
              key: database-url
        - name: GATEWAY_PLUGINS_DIR
          value: "/plugins"
        volumeMounts:
        - name: plugins
          mountPath: /plugins
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "2000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 3000
          initialDelaySeconds: 30
          periodSeconds: 10
      volumes:
      - name: plugins
        persistentVolumeClaim:
          claimName: openclaw-plugins
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: openclaw-config
data:
  gateway.yaml: |
    gateway:
      bind: "0.0.0.0:3000"
      remote:
        url: "https://gateway.tropical-media.work"
      plugins:
        entries:
          - name: n8n-integration
            path: /plugins/n8n-integration
            config:
              webhookUrl: "https://n8n.tropical-media.work/webhook"
          - name: model-router
            path: /plugins/model-router
            config:
              defaultModel: "gpt-4o"
              fallbackModel: "gpt-4o-mini"
              costOptimization: true
      security:
        maxRequestSize: "10MB"
        rateLimitPerMinute: 100
        allowedOrigins:
          - "https://tropical-media.work"
          - "https://*.tropical-media.work"