Docker Scenarios LLD Deep Dive

flowchart LR
    User["Web / Mobile Chat Client"] --> Edge["CloudFront + ALB / API Gateway"]
    Edge --> Fargate["ECS Fargate Orchestrator Task"]
    Edge --> Lambda["Lambda Burst Worker"]
    Fargate --> L1["L1 In-Memory Cache"]
    Fargate --> Redis["Redis L2 Cache"]
    Fargate --> DAX["DAX / DynamoDB Conversation Path"]
    Fargate --> ECR["Image from ECR"]

# What a SINGLE-STAGE image looks like internally
├── gcc 12.2          (~150 MB)      ← Never used at runtime
├── python3-dev       (~45 MB)       ← Only needed for C extensions during pip install
├── pip cache         (~200 MB)      ← Leftover .whl files
├── test frameworks   (~30 MB)       ← pytest, coverage, moto
├── .git directory    (~80 MB)       ← Accidentally copied
├── node_modules/     (~180 MB)      ← Build tool artifacts
├── YOUR ACTUAL APP   (~25 MB)       ← What you actually need

# Stage 1: Build
FROM python:3.11-slim AS builder
COPY requirements.txt .
RUN pip install --user -r requirements.txt
COPY . .
RUN python setup.py build

# Stage 2: Runtime
FROM python:3.11-slim AS runtime
COPY --from=builder /root/.local /root/.local
COPY --from=builder /app/dist /app
EXPOSE 8080
CMD ["python", "-m", "app.main"]

# Single-flight pattern for cache reads
import asyncio
from functools import lru_cache

class SingleFlightCache:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.l1 = {}
        self.in_flight = {}  # key -> asyncio.Future

    async def get(self, key):
        # L1 check
        if key in self.l1:
            return self.l1[key]

        # Single-flight: if someone else is already fetching, wait for them
        if key in self.in_flight:
            return await self.in_flight[key]

        # I'm the first — create a flight
        future = asyncio.get_event_loop().create_future()
        self.in_flight[key] = future

        try:
            value = await self.redis.get(key)
            if value is None:
                value = await self.fetch_from_dynamodb(key)
                await self.redis.setex(key, 300, value)  # 5 min TTL

            # Populate L1 with jittered TTL
            import random
            ttl = 15 + random.randint(0, 10)  # 15-25 seconds
            self.l1[key] = value
            asyncio.get_event_loop().call_later(ttl, self.l1.pop, key, None)

            future.set_result(value)
            return value
        finally:
            del self.in_flight[key]

# WRONG — secret leaks into image layer
ARG NPM_TOKEN
RUN echo "//registry.npmjs.org/:_authToken=${NPM_TOKEN}" > .npmrc && \
    npm install && \
    rm .npmrc  # ← STILL IN THE LAYER! 'docker history' reveals it

# RIGHT — BuildKit secret mount, never persists
# syntax=docker/dockerfile:1
RUN --mount=type=secret,id=npm_token \
    NPM_TOKEN=$(cat /run/secrets/npm_token) \
    echo "//registry.npmjs.org/:_authToken=${NPM_TOKEN}" > .npmrc && \
    npm install && \
    rm .npmrc
# Secret is mounted during the RUN step only — never part of any layer

Second 0:   New task definition deployed via CodeDeploy Blue/Green
Second 0-60: Canary phase — 5% traffic routed to new tasks
Second 45:  CloudWatch alarm triggers: P99 latency > 500ms on new tasks
Second 46:  CodeDeploy automatic rollback initiated
Second 46-90: Old task set receives 100% traffic (blue/green makes this instant)
Second 90:  New tasks drained and stopped
Second 91:  Rollback complete. Alert sent to Slack with deployment ID and alarm details.
Total user impact: ~45 seconds of 5% traffic seeing degraded latency.

# ============================================
# PRODUCTION MULTI-STAGE DOCKERFILE
# Orchestrator Service for Chat Platform
# ============================================

# ---------- Stage 1: Dependency Installation ----------
FROM python:3.11.7-slim-bookworm AS deps

WORKDIR /app

# Install system dependencies needed for C extensions
RUN apt-get update && \
    apt-get install -y --no-install-recommends gcc libpq-dev && \
    rm -rf /var/lib/apt/lists/*

# Copy ONLY dependency files first — layer caching optimization
# If requirements.txt hasn't changed, this layer is cached
COPY requirements.txt requirements-lock.txt ./

# Install Python dependencies into a user directory (easy to copy later)
RUN pip install --no-cache-dir --user \
    -r requirements-lock.txt \
    --require-hashes

# ---------- Stage 2: Test Runner (CI quality gate) ----------
FROM deps AS test

COPY requirements-test.txt ./
RUN pip install --no-cache-dir --user -r requirements-test.txt

COPY . .
RUN python -m pytest tests/unit \
    --tb=short \
    --no-header \
    -q

# ---------- Stage 3: Production Runtime ----------
FROM python:3.11.7-slim-bookworm AS runtime

# Security: run as non-root
RUN groupadd -r appuser && \
    useradd -r -g appuser -d /app -s /sbin/nologin appuser

WORKDIR /app

# Copy ONLY the installed packages from deps stage (not gcc, not pip cache)
COPY --from=deps /root/.local /home/appuser/.local

# Copy application code
COPY --chown=appuser:appuser src/ ./src/
COPY --chown=appuser:appuser config/ ./config/

# Set PATH for user-installed packages
ENV PATH="/home/appuser/.local/bin:${PATH}" \
    PYTHONPATH="/app/src" \
    PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1

USER appuser

# Health check — ALB and ECS use this
HEALTHCHECK --interval=15s --timeout=5s --start-period=30s --retries=3 \
    CMD ["python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8080/health')"]

EXPOSE 8080

ENTRYPOINT ["python", "-m", "uvicorn"]
CMD ["src.main:app", "--host", "0.0.0.0", "--port", "8080", "--workers", "4"]

# ============================================
# ECS Fargate Task Definition (CloudFormation snippet)
# ============================================
OrchestratorTaskDefinition:
  Type: AWS::ECS::TaskDefinition
  Properties:
    Family: chatbot-orchestrator
    Cpu: "1024"          # 1 vCPU
    Memory: "2048"       # 2 GB
    NetworkMode: awsvpc
    RequiresCompatibilities: [FARGATE]
    ExecutionRoleArn: !GetAtt ExecutionRole.Arn
    TaskRoleArn: !GetAtt TaskRole.Arn
    ContainerDefinitions:
      - Name: orchestrator
        Image: !Sub "${AWS::AccountId}.dkr.ecr.${AWS::Region}.amazonaws.com/chatbot-orchestrator:${ImageTag}"
        PortMappings:
          - ContainerPort: 8080
            Protocol: tcp
        LogConfiguration:
          LogDriver: awslogs
          Options:
            awslogs-group: /ecs/chatbot-orchestrator
            awslogs-region: !Ref AWS::Region
            awslogs-stream-prefix: ecs
        HealthCheck:
          Command: ["CMD-SHELL", "python -c \"import urllib.request; urllib.request.urlopen('http://localhost:8080/health')\" || exit 1"]
          Interval: 15
          Timeout: 5
          Retries: 3
          StartPeriod: 30
        Environment:
          - Name: REDIS_ENDPOINT
            Value: !GetAtt RedisCluster.PrimaryEndPoint.Address
          - Name: DYNAMODB_TABLE
            Value: !Ref ConversationTable
          - Name: LOG_LEVEL
            Value: INFO

╔══════════════════════════════════════════════════════════════════════╗
║                    CRITICAL POINTS — LLD-1                         ║
╠══════════════════════════════════════════════════════════════════════╣
║                                                                      ║
║  1. COMPUTE CHOICE = f(service_count, traffic_shape, team_size)    ║
║     • <10 services → Fargate wins over EKS                        ║
║     • Streaming/WebSocket → Fargate wins over Lambda               ║
║     • Cache locality needed → Fargate wins over Lambda             ║
║                                                                      ║
║  2. MULTI-STAGE IS NOT JUST ABOUT SIZE                             ║
║     • Size → faster pull → faster scale-out → faster rollback      ║
║     • Attack surface → fewer packages → fewer CVEs → cleaner scans ║
║     • Build reproducibility → deterministic runtime image           ║
║                                                                      ║
║  3. CACHE HIERARCHY IS A LATENCY PYRAMID                           ║
║     • L1 (container memory): <1ms, 60-70% hit rate, per-task      ║
║     • L2 (Redis): 1-5ms, shared across tasks, 5-10 min TTL        ║
║     • L3 (DAX/DynamoDB): 5-10ms, persistent, source of truth      ║
║     • NEVER let L1 be the source of truth — it dies with the task  ║
║                                                                      ║
║  4. HYBRID COMPUTE = BASELINE + BURST                              ║
║     • Fargate for baseline: predictable, warm, cached              ║
║     • Lambda for burst: instant scale, no cache, acceptable delay  ║
║     • NEVER run stateful workloads on Lambda                       ║
║                                                                      ║
║  5. DOCKERFILE LAYER ORDER MATTERS                                 ║
║     • Least-changing layers FIRST (OS packages, dependencies)      ║
║     • Most-changing layers LAST (application code)                 ║
║     • One changed layer invalidates ALL layers below it            ║
║                                                                      ║
║  6. SECRETS NEVER IN IMAGES                                        ║
║     • Not in ENV (persists in layers)                              ║
║     • Not in ARG (visible in docker history)                       ║
║     • Use BuildKit --mount=type=secret or runtime injection        ║
║                                                                      ║
╚══════════════════════════════════════════════════════════════════════╝

sequenceDiagram
    participant SM as SageMaker
    participant C as Inference Container
    participant W as Warmup Script
    participant H as /ping Health Check
    participant LB as Endpoint Load Balancer

    SM->>C: Start container
    C->>W: Run startup warmup
    W->>C: Load weights, trigger JIT, cache CUDA kernels
    SM->>H: Probe /ping
    H-->>SM: 503 warming_up
    W->>C: Warmup complete
    SM->>H: Probe /ping
    H-->>SM: 200 healthy
    SM->>LB: Add instance to live traffic pool

Container Lifecycle for ML Inference:

State 1: PROCESS ALIVE (seconds 0-5)
├── Python process started
├── HTTP server listening on port 8080
├── Docker HEALTHCHECK: PASS ✓
├── Can serve inference: NO ✗
└── What's missing: model weights, CUDA context, JIT kernels

State 2: MODEL LOADED (seconds 5-60)
├── Weights loaded into GPU VRAM
├── CUDA context initialized
├── JIT kernels NOT yet compiled
├── First real request: SLOW (3-5x normal latency)
└── What's missing: warmed kernel cache, primed memory allocator

State 3: TRAFFIC READY (seconds 60-120)
├── Synthetic warmup requests completed
├── Short/medium/long prompt shapes all exercised
├── CUDA kernel cache hot
├── Memory allocator has seen realistic allocation patterns
├── /ping returns 200
└── Load balancer routes traffic: NOW SAFE

Scenario: MinCapacity = 1
- Instance fails or is replaced by SageMaker
- Time to new instance ready: ~120 seconds
- During those 120 seconds: 100% of traffic hits cold path
- If traffic = 50 requests/sec → 6,000 requests see degraded latency
- Each degraded request adds ~3 seconds → user-visible SLA violation

Scenario: MinCapacity = 2
- Instance fails or is replaced by SageMaker
- Surviving instance handles ALL traffic immediately
- New instance warms up in background
- Zero requests see cold-path latency
- Extra cost: ~$2-4/hour for one additional GPU instance
- SLA violations prevented: priceless (or rather, easily worth $2-4/hour)

# Warmup with timeout and failure handling
import asyncio
import signal
import sys
import logging

logger = logging.getLogger("warmup")

WARMUP_TIMEOUT_SECONDS = 180
WARMUP_SHAPES = [
    {"prompt": "Hello", "max_tokens": 10, "label": "short"},
    {"prompt": "Explain the plot of this manga chapter in detail, covering all character interactions and themes discussed",
     "max_tokens": 200, "label": "medium"},
    {"prompt": "You are a manga expert assistant..." + " context " * 200,
     "max_tokens": 512, "label": "long"},
]

class WarmupState:
    ready = False
    started_at = None
    completed_shapes = set()

warmup_state = WarmupState()

async def run_warmup(model):
    """Execute warmup with timeout protection."""
    warmup_state.started_at = asyncio.get_event_loop().time()

    try:
        async with asyncio.timeout(WARMUP_TIMEOUT_SECONDS):
            for shape in WARMUP_SHAPES:
                logger.info(f"Warming shape: {shape['label']}")
                try:
                    await model.generate(
                        prompt=shape["prompt"],
                        max_tokens=shape["max_tokens"]
                    )
                    warmup_state.completed_shapes.add(shape["label"])
                    logger.info(f"Shape {shape['label']} warmed successfully")
                except Exception as e:
                    logger.error(f"Warmup failed for {shape['label']}: {e}")
                    # Fail the entire warmup if any shape fails
                    sys.exit(1)

            warmup_state.ready = True
            elapsed = asyncio.get_event_loop().time() - warmup_state.started_at
            logger.info(f"All shapes warmed in {elapsed:.1f}s")

    except asyncio.TimeoutError:
        logger.critical(
            f"Warmup timed out after {WARMUP_TIMEOUT_SECONDS}s. "
            f"Completed: {warmup_state.completed_shapes}. Exiting."
        )
        sys.exit(1)  # SageMaker will replace this container

# Health check endpoint
async def ping_handler(request):
    """SageMaker probes this. 503 = not ready. 200 = serve traffic."""
    if warmup_state.ready:
        return web.json_response({"status": "healthy"}, status=200)
    else:
        return web.json_response(
            {"status": "warming_up",
             "completed": list(warmup_state.completed_shapes)},
            status=503
        )

# SageMaker blue/green model deployment
import boto3

sm = boto3.client("sagemaker")

def deploy_new_model(endpoint_name, new_model_name, new_variant_weight=0.05):
    """
    Step 1: Add new model variant at 5% weight (canary)
    Step 2: Monitor for 15 minutes
    Step 3: Shift to 100% or rollback
    """

    # Create new model
    sm.create_model(
        ModelName=new_model_name,
        PrimaryContainer={
            "Image": "123456789.dkr.ecr.us-east-1.amazonaws.com/inference:v2.1",
            "ModelDataUrl": "s3://models/v2.1/model.tar.gz",
        },
        ExecutionRoleArn="arn:aws:iam::role/SageMakerRole"
    )

    # Create new endpoint config with canary traffic split
    sm.create_endpoint_config(
        EndpointConfigName=f"{endpoint_name}-canary",
        ProductionVariants=[
            {
                "VariantName": "current",
                "ModelName": "model-v2.0",
                "InstanceType": "ml.g5.2xlarge",
                "InitialInstanceCount": 2,
                "InitialVariantWeight": 1.0 - new_variant_weight,
            },
            {
                "VariantName": "canary",
                "ModelName": new_model_name,
                "InstanceType": "ml.g5.2xlarge",
                "InitialInstanceCount": 2,  # Both instances warm before traffic
                "InitialVariantWeight": new_variant_weight,
            }
        ]
    )

    # Update endpoint — SageMaker handles rolling replacement
    sm.update_endpoint(
        EndpointName=endpoint_name,
        EndpointConfigName=f"{endpoint_name}-canary",
    )
    # New instances boot → warmup runs → /ping returns 200 → traffic shifts

#!/usr/bin/env python3
"""
SageMaker inference container entrypoint.
Handles model loading, warmup, health checks, and inference serving.
"""

import os
import time
import logging
from pathlib import Path
from aiohttp import web
from safetensors.torch import load_file

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("inference-container")

# ============================================
# MODEL LOADING WITH INTEGRITY CHECK
# ============================================
class ModelServer:
    def __init__(self):
        self.model = None
        self.tokenizer = None
        self.ready = False
        self.load_time = None
        self.warmup_time = None

    def load_model(self, model_dir="/opt/ml/model"):
        """Load model with safetensors integrity check."""
        start = time.monotonic()

        model_path = Path(model_dir)
        safetensor_files = list(model_path.glob("*.safetensors"))

        if not safetensor_files:
            raise FileNotFoundError(f"No .safetensors files in {model_dir}")

        logger.info(f"Loading {len(safetensor_files)} safetensor shards...")

        # safetensors validates integrity on load — corrupted files throw here
        try:
            state_dict = {}
            for f in safetensor_files:
                state_dict.update(load_file(str(f), device="cuda"))
            self.model = self._build_model(state_dict)
            self.load_time = time.monotonic() - start
            logger.info(f"Model loaded in {self.load_time:.1f}s")
        except Exception as e:
            logger.critical(f"Model load failed: {e}")
            raise

    async def warmup(self):
        """Warmup with representative prompt shapes."""
        start = time.monotonic()
        shapes = [
            ("short", "Hello", 10),
            ("medium", "Explain this chapter thoroughly " * 10, 200),
            ("long", "System: You are an expert... " * 50, 512),
        ]
        for label, prompt, max_tokens in shapes:
            logger.info(f"Warmup: {label} ({len(prompt)} chars, {max_tokens} max)")
            await self.generate(prompt, max_tokens)
            logger.info(f"Warmup: {label} complete")

        self.warmup_time = time.monotonic() - start
        self.ready = True
        logger.info(f"Warmup done in {self.warmup_time:.1f}s. READY FOR TRAFFIC.")

# ============================================
# HEALTH CHECK — THE READINESS GATE
# ============================================
async def health_check(request):
    server = request.app["model_server"]
    if server.ready:
        return web.json_response({
            "status": "healthy",
            "load_time_s": server.load_time,
            "warmup_time_s": server.warmup_time,
        }, status=200)
    return web.json_response({"status": "warming_up"}, status=503)

async def invocations(request):
    """SageMaker sends inference requests to /invocations."""
    server = request.app["model_server"]
    if not server.ready:
        return web.json_response({"error": "Model not ready"}, status=503)
    body = await request.json()
    result = await server.generate(body["prompt"], body.get("max_tokens", 256))
    return web.json_response({"generated_text": result})

# ============================================
# STARTUP SEQUENCE
# ============================================
async def startup(app):
    server = ModelServer()
    app["model_server"] = server
    server.load_model()       # Blocks until weights are in GPU
    await server.warmup()     # Blocks until all shapes are warmed

app = web.Application()
app.on_startup.append(startup)
app.router.add_get("/ping", health_check)
app.router.add_post("/invocations", invocations)

if __name__ == "__main__":
    web.run_app(app, host="0.0.0.0", port=8080)

╔══════════════════════════════════════════════════════════════════════╗
║                    CRITICAL POINTS — LLD-2                         ║
╠══════════════════════════════════════════════════════════════════════╣
║                                                                      ║
║  1. THREE CONTAINER STATES (memorize these)                        ║
║     • PROCESS ALIVE ≠ MODEL LOADED ≠ TRAFFIC READY                ║
║     • Docker sees alive. LB needs ready. Users need warm.          ║
║     • If you conflate these, users hit cold containers.            ║
║                                                                      ║
║  2. READINESS GATE = /ping RETURNS 503 UNTIL WARM                 ║
║     • The health check IS the admission control                    ║
║     • 503 during warmup is CORRECT behavior, not an error          ║
║     • Never return 200 just because the process started            ║
║                                                                      ║
║  3. MinCapacity = 2 IS INSURANCE, NOT WASTE                       ║
║     • One instance fails → other absorbs 100% immediately          ║
║     • Cost of extra instance << cost of cold-start SLA violations  ║
║     • The math always favors MinCapacity ≥ 2 for production        ║
║                                                                      ║
║  4. WARMUP MUST MATCH PRODUCTION SHAPES                            ║
║     • Short + medium + long prompt shapes                          ║
║     • CUDA kernel compilation is shape-dependent                   ║
║     • Warming with "hello" does NOT warm 500-token prompts         ║
║                                                                      ║
║  5. SAFETENSORS > PYTORCH .BIN (always)                            ║
║     • 2-5x faster loading (memory-mapped I/O)                     ║
║     • No arbitrary code execution (pickle is dangerous)            ║
║     • Built-in integrity validation                                ║
║                                                                      ║
║  6. ADD WARMUP TIMEOUT — DEADLOCKED WARMUP IS SILENT DEATH        ║
║     • Set deadline (e.g., 180 seconds)                             ║
║     • Exit with code 1 on timeout → container replaced             ║
║     • Alarm on repeated timeouts → systemic issue                  ║
║                                                                      ║
║  7. ROLLING DEPLOYMENT FOR MODEL UPDATES                           ║
║     • Never update all instances simultaneously                    ║
║     • Blue/green at SageMaker level                                ║
║     • New fleet warms → passes readiness → receives traffic        ║
║                                                                      ║
╚══════════════════════════════════════════════════════════════════════╝

flowchart LR
    Req["Chat Request"] --> API["OpenAI-Compatible API Layer"]
    API --> Sched["Continuous Batching Scheduler"]
    Sched --> Prefix["Prefix Cache"]
    Prefix --> KV["PagedAttention KV Block Manager"]
    KV --> Engine["vLLM Engine"]
    Engine --> GPU["GPU"]
    Engine --> LoRA["Optional Multi-LoRA Adapter"]

Traditional KV Cache Allocation:
Request A: [████████████████████░░░░░░░░░░░░░░░░░░░░] ← 50% wasted
Request B: [██████████████░░░░░░░░░░░░░░░░░░░░░░░░░░] ← 65% wasted
Request C: [████████████████████████████░░░░░░░░░░░░░] ← 30% wasted
           ^^^^^ USED ^^^^^  ^^^^^^^^ EMPTY/WASTED ^^^^^^^^

PagedAttention KV Cache:
Request A: [page][page][page][page][page]     ← 5 pages, fits exactly
Request B: [page][page][page]                 ← 3 pages, fits exactly
Request C: [page][page][page][page][page][page][page]  ← 7 pages, fits exactly
Free pool: [page][page][page][page][page]...  ← available for new requests
           ^^^^^ Only allocates what's needed ^^^^^

Static Batching (old way):
Time ──────────────────────────────────────────►
Batch 1: [Req A ████████] [Req B ██████████████] [Req C ████]
         ^--- All three start together, all must finish before next batch ---^
         Req A finishes early but GPU slot is WASTED until Req B finishes
         New requests WAIT until entire batch completes

Continuous Batching (vLLM):
Time ──────────────────────────────────────────►
Slot 1:  [Req A ████████] [Req D ██████] [Req G ████████████]
Slot 2:  [Req B ██████████████] [Req E ████████████]
Slot 3:  [Req C ████] [Req F ██████████] [Req H ██████]
         ^--- Requests enter/exit at iteration boundaries ---^
         When Req A finishes, Req D immediately starts in that slot
         GPU never idles waiting for the longest request in a batch

# vLLM server launch configuration
# This is the actual Docker CMD / entrypoint for our inference container

"""
vLLM Serving Container Launch Script
Run inside: FROM vllm/vllm-openai:v0.4.0
"""
import subprocess
import os

VLLM_ARGS = [
    "python", "-m", "vllm.entrypoints.openai.api_server",

    # Model identity
    "--model", os.environ.get("MODEL_PATH", "/opt/ml/model"),
    "--served-model-name", os.environ.get("MODEL_NAME", "chatbot-v2"),

    # Memory management
    "--gpu-memory-utilization", "0.92",       # Leave 8% CUDA headroom
    "--max-model-len", "4096",                # Context window cap
    "--block-size", "16",                     # PagedAttention block size

    # Concurrency control
    "--max-num-seqs", "128",                  # Max concurrent sequences
    "--max-num-batched-tokens", "8192",       # Max tokens per batch iteration

    # Performance optimizations
    "--enable-prefix-caching",                # Reuse common prompt prefixes
    "--quantization", "awq",                  # INT4 quantization
    "--dtype", "half",                        # FP16 for non-quantized layers

    # Serving config
    "--host", "0.0.0.0",
    "--port", "8080",
    "--trust-remote-code",

    # Tensor parallelism (for multi-GPU setups)
    "--tensor-parallel-size", os.environ.get("TP_SIZE", "1"),
]

# Optional: Multi-LoRA support
if os.environ.get("ENABLE_LORA", "false") == "true":
    VLLM_ARGS.extend([
        "--enable-lora",
        "--max-loras", os.environ.get("MAX_LORAS", "4"),
        "--max-lora-rank", os.environ.get("MAX_LORA_RANK", "16"),
    ])

print(f"Starting vLLM with args: {' '.join(VLLM_ARGS)}")
subprocess.run(VLLM_ARGS, check=True)

# Multi-LoRA request routing — client-side
import openai

# All adapters are served from the SAME container on the SAME GPU
client = openai.OpenAI(base_url="http://inference:8080/v1", api_key="dummy")

# Japanese adapter
response_ja = client.chat.completions.create(
    model="chatbot-ja",       # ← Routes to Japanese LoRA adapter
    messages=[{"role": "user", "content": "この漫画の感想を教えて"}],
    max_tokens=256,
)

# English adapter — same container, same GPU, different adapter
response_en = client.chat.completions.create(
    model="chatbot-en",       # ← Routes to English LoRA adapter
    messages=[{"role": "user", "content": "Tell me about this manga"}],
    max_tokens=256,
)

# Both use the same base model weights — only adapter layers differ

# Prometheus scrape config for vLLM monitoring
# Added to the inference container sidecar
scrape_configs:
  - job_name: 'vllm-inference'
    metrics_path: '/metrics'
    static_configs:
      - targets: ['localhost:8080']
    scrape_interval: 15s

# CloudWatch alarm definitions
alarms:
  kv_cache_pressure:
    metric: vllm_gpu_cache_usage_perc
    threshold: 0.95
    period: 300        # 5 minutes
    evaluation_periods: 2
    action: scale_out_gpu_fleet

  preemption_rate:
    metric: vllm_num_preemptions_total
    threshold: 10      # 10 preemptions in 5 minutes
    period: 300
    action: page_on_call

  request_queue_depth:
    metric: vllm_num_requests_waiting
    threshold: 50
    period: 60
    action: scale_out_gpu_fleet

# ============================================
# vLLM INFERENCE CONTAINER
# Production-ready GPU serving container
# ============================================

FROM nvidia/cuda:12.1.1-runtime-ubuntu22.04 AS base

# Avoid timezone prompts
ENV DEBIAN_FRONTEND=noninteractive

RUN apt-get update && \
    apt-get install -y --no-install-recommends \
    python3.11 python3-pip curl && \
    rm -rf /var/lib/apt/lists/*

# ---------- Stage: Dependencies ----------
FROM base AS deps

# Install vLLM and serving dependencies
COPY requirements-inference.txt /tmp/
RUN pip install --no-cache-dir -r /tmp/requirements-inference.txt

# ---------- Stage: Runtime ----------
FROM base AS runtime

# Copy installed packages
COPY --from=deps /usr/local/lib/python3.11 /usr/local/lib/python3.11
COPY --from=deps /usr/local/bin /usr/local/bin

# Copy serving scripts
WORKDIR /app
COPY serve.py warmup.py health.py ./
COPY config/ ./config/

# Non-root user (as much as possible — some GPU ops need root)
RUN useradd -m -s /bin/bash vllm
USER vllm

# vLLM metrics endpoint + inference API
EXPOSE 8080

# Health check for SageMaker
HEALTHCHECK --interval=10s --timeout=5s --start-period=300s --retries=3 \
    CMD curl -f http://localhost:8080/health || exit 1

ENV NVIDIA_VISIBLE_DEVICES=all \
    NVIDIA_DRIVER_CAPABILITIES=compute,utility \
    VLLM_USAGE_STATS=0

ENTRYPOINT ["python3", "serve.py"]

╔══════════════════════════════════════════════════════════════════════╗
║                    CRITICAL POINTS — LLD-3                         ║
╠══════════════════════════════════════════════════════════════════════╣
║                                                                      ║
║  1. PagedAttention = OS VIRTUAL MEMORY FOR GPU                     ║
║     • Static KV = contiguous allocation → 72% waste                ║
║     • Paged KV = on-demand blocks → 4% waste                      ║
║     • This single change increased concurrency 15x                 ║
║                                                                      ║
║  2. CONTINUOUS BATCHING = NO WAITING FOR SLOWEST REQUEST           ║
║     • Static batch: all wait for longest → GPU idle 40-60%         ║
║     • Continuous: enter/exit at iteration boundary → 85-95% util   ║
║     • Critical for chat workloads with variable response lengths   ║
║                                                                      ║
║  3. PREFIX CACHING = FREE COMPUTE SAVINGS FOR CHAT                 ║
║     • System prompt is identical across ALL requests               ║
║     • Without caching: recompute 500 tokens every time             ║
║     • With caching: reuse computed KV cache → saves 500ms+         ║
║                                                                      ║
║  4. gpu_memory_utilization = 0.92 (NOT 1.0, NOT 0.80)             ║
║     • 1.0 → sporadic CUDA OOM (no headroom for CUDA internals)    ║
║     • 0.80 → wasting 16 GB of usable KV cache space               ║
║     • Profile under peak load, validate with soak test             ║
║                                                                      ║
║  5. vLLM vs TensorRT-LLM = OPERABILITY vs RAW SPEED               ║
║     • TensorRT-LLM: 10-20% faster, 3-5x harder to operate        ║
║     • vLLM: near-best speed, pip install, broad hardware support   ║
║     • Choose based on team skills, not benchmarks                  ║
║                                                                      ║
║  6. MULTI-LoRA = MULTIPLE MODELS ON ONE GPU                        ║
║     • Base model loaded once → shared across all adapters          ║
║     • Each adapter: 0.1-1% memory overhead                         ║
║     • Route by model name in OpenAI-compatible API format          ║
║                                                                      ║
║  7. MONITOR THESE 5 vLLM METRICS                                   ║
║     • num_requests_running (concurrency)                           ║
║     • gpu_cache_usage_perc (KV pressure)                           ║
║     • num_preemptions_total (capacity overflow)                    ║
║     • avg_prompt_throughput (prefill speed)                        ║
║     • avg_generation_throughput (decode speed)                     ║
║                                                                      ║
╚══════════════════════════════════════════════════════════════════════╝

flowchart TD
    Msg["Conversation History"] --> Budget["Context Budget Manager"]
    Budget --> Window["Sliding Window Selection"]
    Window --> Prompt["Prompt Assembly"]
    Prompt --> Infer["vLLM Inference"]
    Infer --> OOM{"OOM?"}
    OOM -->|No| Resp["Return Response"]
    OOM -->|Yes| Guard["OOM Circuit Breaker"]
    Guard --> Fallback["Graceful Fallback + Metric"]

Turn  1: [SYS:500][Q1:50][A1:200]                             = 750 tokens   ✓ OK
Turn  5: [SYS:500][Q1:50][A1:200]...[Q5:50][A5:200]           = 2,250 tokens ✓ OK
Turn 10: [SYS:500][Q1:50][A1:200]...[Q10:50][A10:200]         = 4,000 tokens ⚠ Getting tight
Turn 20: [SYS:500][Q1:50][A1:200]...[Q20:50][A20:200]         = 7,500 tokens ⚠ VRAM pressure
Turn 30: [SYS:500][Q1:50][A1:200]...[Q30:50][A30:200]         = 11,000 tokens ✗ GPU OOM!

# Simplified but accurate representation of the budget logic

MAX_CONTEXT = 4096  # Model's max context window

def assemble_prompt(system_prompt, rag_context, history, current_query):
    budget = MAX_CONTEXT

    # 1. System prompt: ALWAYS included, non-negotiable
    budget -= count_tokens(system_prompt)       # -500 tokens → 3596 left

    # 2. Output reservation: MUST leave room for the response
    budget -= MAX_OUTPUT_TOKENS                 # -512 tokens → 3084 left

    # 3. Current query: ALWAYS included
    budget -= count_tokens(current_query)       # -50 tokens → 3034 left

    # 4. RAG context: Include retrieval results
    budget -= count_tokens(rag_context)         # -300 tokens → 2734 left

    # 5. History: Fill remaining budget with MOST RECENT turns
    history_tokens = 0
    included_turns = []
    for turn in reversed(history):              # Start from most recent
        turn_tokens = count_tokens(turn)
        if history_tokens + turn_tokens > budget:
            break                               # Stop, budget exhausted
        included_turns.insert(0, turn)
        history_tokens += turn_tokens

    # 6. If any history was dropped, add a truncation marker
    if len(included_turns) < len(history):
        included_turns.insert(0,
            "[Earlier messages were summarized to fit context window]")

    return build_prompt(system_prompt, included_turns,
                       rag_context, current_query)

# Enhanced sliding window with pinned turns and entity extraction
from dataclasses import dataclass
from typing import Optional

@dataclass
class ConversationTurn:
    role: str
    content: str
    token_count: int
    is_pinned: bool = False      # First turn, entity-rich turns
    entities: list = None        # Extracted entities for compaction

class ContextBudgetManager:
    def __init__(self, max_context=4096, output_reserve=512,
                 system_reserve=500, rag_reserve=300):
        self.max_context = max_context
        self.output_reserve = output_reserve
        self.system_reserve = system_reserve
        self.rag_reserve = rag_reserve

    def assemble(self, system_prompt: str, rag_context: str,
                 current_query: str, history: list[ConversationTurn]) -> dict:
        """Build prompt within budget. Never exceeds max_context."""

        budget = self.max_context
        budget -= self.system_reserve
        budget -= self.output_reserve
        budget -= self._count_tokens(current_query)
        budget -= self._count_tokens(rag_context)

        # Phase 1: Always include pinned turns
        pinned = [t for t in history if t.is_pinned]
        unpinned = [t for t in history if not t.is_pinned]

        pinned_cost = sum(t.token_count for t in pinned)
        remaining_budget = budget - pinned_cost

        if remaining_budget < 0:
            # Even pinned turns don't fit — compact them to entities
            entity_summary = self._compact_to_entities(pinned)
            pinned = [ConversationTurn(
                role="system",
                content=f"[Context from earlier: {entity_summary}]",
                token_count=self._count_tokens(entity_summary),
                is_pinned=True
            )]
            pinned_cost = pinned[0].token_count
            remaining_budget = budget - pinned_cost

        # Phase 2: Fill remaining budget with most recent unpinned turns
        included_unpinned = []
        used = 0
        for turn in reversed(unpinned):
            if used + turn.token_count > remaining_budget:
                break
            included_unpinned.insert(0, turn)
            used += turn.token_count

        # Phase 3: Build final prompt with truncation marker if needed
        total_history = len(pinned) + len(unpinned)
        included_history = len(pinned) + len(included_unpinned)
        was_truncated = included_history < total_history

        result = {
            "system_prompt": system_prompt,
            "history": pinned + (
                [ConversationTurn(
                    role="system",
                    content="[Some earlier messages were omitted to fit context window]",
                    token_count=15
                )] if was_truncated else []
            ) + included_unpinned,
            "rag_context": rag_context,
            "current_query": current_query,
            "was_truncated": was_truncated,
            "turns_dropped": total_history - included_history,
            "total_tokens": self.max_context - remaining_budget + used,
        }
        return result

    def _compact_to_entities(self, turns):
        """Extract key entities from turns for compact representation."""
        entities = []
        for turn in turns:
            if turn.entities:
                entities.extend(turn.entities)
        return ", ".join(entities) if entities else "No key entities extracted"

    def _count_tokens(self, text):
        # Simplified — use tiktoken in production
        return len(text.split()) * 1.3  # rough approximation

# OOM Circuit Breaker implementation
import torch
import time
import logging
from enum import Enum

logger = logging.getLogger("oom-breaker")

class BreakerState(Enum):
    CLOSED = "closed"        # Normal operation
    OPEN = "open"            # Blocking full-context requests
    HALF_OPEN = "half_open"  # Testing recovery

class OOMCircuitBreaker:
    def __init__(self, cooldown_seconds=30, reduced_context=1024,
                 max_consecutive_trips=2, trip_window_seconds=300):
        self.state = BreakerState.CLOSED
        self.cooldown_seconds = cooldown_seconds
        self.reduced_context = reduced_context
        self.max_consecutive_trips = max_consecutive_trips
        self.trip_window_seconds = trip_window_seconds
        self.last_trip_time = 0
        self.trip_count = 0
        self.metrics = {"trips": 0, "reduced_requests": 0, "recovered": 0}

    def get_allowed_context_length(self, requested_length: int) -> int:
        """Returns the maximum context length allowed given breaker state."""
        if self.state == BreakerState.CLOSED:
            return requested_length
        elif self.state == BreakerState.OPEN:
            self.metrics["reduced_requests"] += 1
            return min(requested_length, self.reduced_context)
        elif self.state == BreakerState.HALF_OPEN:
            return requested_length  # Let one full request through to test

    def record_success(self):
        """Called after successful inference."""
        if self.state == BreakerState.HALF_OPEN:
            logger.info("Half-open request succeeded. Closing breaker.")
            self.state = BreakerState.CLOSED
            self.trip_count = 0
            self.metrics["recovered"] += 1

    def record_oom(self, engine):
        """Called when CUDA OOM is caught."""
        now = time.monotonic()
        self.metrics["trips"] += 1

        # Track consecutive trips
        if now - self.last_trip_time < self.trip_window_seconds:
            self.trip_count += 1
        else:
            self.trip_count = 1
        self.last_trip_time = now

        logger.error(
            f"OOM Circuit Breaker TRIPPED (trip #{self.trip_count}). "
            f"Entering OPEN state for {self.cooldown_seconds}s."
        )

        # Emergency GPU cleanup
        torch.cuda.empty_cache()
        engine.reset()  # Flush KV blocks, clear scheduler

        self.state = BreakerState.OPEN

        # Schedule transition to half-open
        # In production, use asyncio.call_later or similar
        self._schedule_half_open()

        # Escalation: too many trips → scale out
        if self.trip_count >= self.max_consecutive_trips:
            logger.critical(
                f"{self.trip_count} OOM trips in {self.trip_window_seconds}s. "
                "Triggering scale-out alarm."
            )
            self._emit_scale_out_alarm()

    def _schedule_half_open(self):
        """After cooldown, allow one test request through."""
        import threading
        def _transition():
            self.state = BreakerState.HALF_OPEN
            logger.info("Breaker entering HALF_OPEN state. Next request is a test.")
        threading.Timer(self.cooldown_seconds, _transition).start()

    def _emit_scale_out_alarm(self):
        """Send CloudWatch metric that triggers autoscaling."""
        import boto3
        cw = boto3.client("cloudwatch")
        cw.put_metric_data(
            Namespace="InferenceContainer",
            MetricData=[{
                "MetricName": "OOMEscalation",
                "Value": 1,
                "Unit": "Count"
            }]
        )

# Full request handler with budget management + OOM protection
import asyncio
from aiohttp import web

class InferenceHandler:
    def __init__(self, engine, budget_manager, circuit_breaker):
        self.engine = engine
        self.budget = budget_manager
        self.breaker = circuit_breaker

    async def handle_request(self, request):
        body = await request.json()

        # Step 1: Assemble prompt within budget
        assembled = self.budget.assemble(
            system_prompt=body.get("system_prompt", "You are a helpful assistant."),
            rag_context=body.get("rag_context", ""),
            current_query=body["query"],
            history=body.get("history", []),
        )

        # Step 2: Check circuit breaker — may reduce context
        allowed_len = self.breaker.get_allowed_context_length(
            assembled["total_tokens"]
        )

        # Step 3: Inference with OOM protection
        try:
            result = await self.engine.generate(
                prompt=assembled,
                max_tokens=min(body.get("max_tokens", 256), 512),
                max_context=allowed_len,
            )
            self.breaker.record_success()

            return web.json_response({
                "response": result.text,
                "tokens_used": result.tokens_used,
                "was_truncated": assembled["was_truncated"],
                "turns_dropped": assembled["turns_dropped"],
            })

        except torch.cuda.OutOfMemoryError:
            # OOM caught — breaker handles cleanup and state transition
            self.breaker.record_oom(self.engine)

            return web.json_response({
                "response": "I'm experiencing high load. Please try again "
                           "or start a new conversation for best results.",
                "error": "capacity_limit",
                "was_truncated": True,
            }, status=503)

        except Exception as e:
            logger.exception(f"Inference failed: {e}")
            return web.json_response(
                {"error": "internal_error"}, status=500
            )

╔══════════════════════════════════════════════════════════════════════╗
║                    CRITICAL POINTS — LLD-4                         ║
╠══════════════════════════════════════════════════════════════════════╣
║                                                                      ║
║  1. GPU OOM IS A PROMPT-SIZE BUG, NOT A HARDWARE BUG               ║
║     • Unbounded history → linear VRAM growth → guaranteed crash     ║
║     • More VRAM just delays the crash, doesn't fix it              ║
║     • Fix: budget manager that CAPS total tokens                   ║
║                                                                      ║
║  2. TOKEN BUDGET PRIORITY (memorize this order)                    ║
║     1. System prompt (always included)                             ║
║     2. Output reservation (always reserved)                        ║
║     3. Current query (always included)                             ║
║     4. RAG context (always included)                               ║
║     5. Pinned turns (first turn, entity-rich turns)                ║
║     6. Recent history (fills remaining budget)                     ║
║     → History is the FLEXIBLE part. Everything else is fixed.      ║
║                                                                      ║
║  3. CIRCUIT BREAKER STATES                                         ║
║     • CLOSED → normal operation, full context allowed              ║
║     • OPEN → reduced context (1024 max), cooldown active           ║
║     • HALF_OPEN → one full request allowed to test recovery        ║
║     • 2 trips in 5 minutes → auto-scale-out alarm                 ║
║                                                                      ║
║  4. CUDA OOM ≠ SYSTEM OOM                                          ║
║     • System OOM → kernel kills process → exit code 137            ║
║     • CUDA OOM → Python exception → catchable but GPU is dirty     ║
║     • After CUDA OOM: empty_cache() + engine reset + KV flush      ║
║                                                                      ║
║  5. TRUNCATION MUST BE EXPLICIT, NEVER SILENT                      ║
║     • Silent drop → model hallucinates with missing context        ║
║     • Explicit marker → model knows to avoid referencing dropped   ║
║     • User-facing marker → transparency for long conversations     ║
║                                                                      ║
║  6. AWQ CALIBRATION IS DOMAIN-SPECIFIC                             ║
║     • 256 in-domain examples is sufficient                         ║
║     • Wrong calibration data → 3-5% quality degradation            ║
║     • Right calibration data → <1% quality degradation             ║
║                                                                      ║
║  7. THE FORMULA: Container Stability =                             ║
║     Budget Management + Weight Quantization + Error Containment    ║
║     Remove any one → instability returns                           ║
║                                                                      ║
╚══════════════════════════════════════════════════════════════════════╝

flowchart LR
    Suite["CI Test Job"] --> Unit["Unit Tests + Pure Mocks"]
    Suite --> Local["Docker Network"]
    Local --> LS["LocalStack"]
    Local --> Redis["Redis Test Container"]
    Local --> OS["OpenSearch Test Container"]
    Suite --> WM["WireMock"]
    Suite --> SM["Staging SageMaker Endpoint"]

                    /\
                   /  \        E2E Tests (staging env)
                  / E2E\       - 10 minutes
                 /------\      - Full shared environment
                / Docker \     - Run selectively on release branches
               / Integr.  \
              /  Tests      \  Docker Integration Tests
             /--------------\  - 5 minutes
            /    LocalStack   \ - LocalStack, Redis, OpenSearch containers
           /    TestContainers \ - Per-PR, disposable, isolated
          /     WireMock        \
         /----------------------\
        /     Unit Tests          \  Unit Tests (mocks)
       /     (Mocks Only)          \ - 30 seconds
      /____________________________\ - Every commit
                                     - Pure business logic

# docker-compose.test.yml
services:
  localstack:
    image: localstack/localstack
    ports:
      - "4566:4566"        # All AWS services on one port
    environment:
      - SERVICES=dynamodb,s3,sqs,kinesis
      - DEFAULT_REGION=us-east-1

# In your test file
from testcontainers.redis import RedisContainer

def test_cache_ttl_behavior():
    with RedisContainer("redis:7.2") as redis:
        client = redis.get_client()
        client.set("key", "value", ex=2)  # 2-second TTL
        assert client.get("key") == b"value"
        time.sleep(3)
        assert client.get("key") is None  # TTL expired — this is REAL Redis behavior

{
  "mappings": [
    {
      "scenarioName": "retry-test",
      "requiredScenarioState": "Started",
      "newScenarioState": "First-Failure",
      "request": { "method": "POST", "url": "/api/inference" },
      "response": { "status": 503, "fixedDelayMilliseconds": 5000 }
    },
    {
      "scenarioName": "retry-test",
      "requiredScenarioState": "First-Failure",
      "newScenarioState": "Success",
      "request": { "method": "POST", "url": "/api/inference" },
      "response": { "status": 200, "body": "{\"result\": \"ok\"}" }
    }
  ]
}

# Test isolation with unique namespaces per test suite
import pytest
import uuid
import boto3

@pytest.fixture(scope="session")
def dynamodb_table():
    """Create an isolated DynamoDB table for this test session."""
    session_id = uuid.uuid4().hex[:8]
    table_name = f"test_{session_id}_conversations"

    # Connect to LocalStack
    dynamodb = boto3.resource(
        "dynamodb",
        endpoint_url="http://localhost:4566",
        region_name="us-east-1",
        aws_access_key_id="test",
        aws_secret_access_key="test",
    )

    # Create isolated table
    table = dynamodb.create_table(
        TableName=table_name,
        KeySchema=[
            {"AttributeName": "conversation_id", "KeyType": "HASH"},
            {"AttributeName": "turn_id", "KeyType": "RANGE"},
        ],
        AttributeDefinitions=[
            {"AttributeName": "conversation_id", "AttributeType": "S"},
            {"AttributeName": "turn_id", "AttributeType": "N"},
        ],
        BillingMode="PAY_PER_REQUEST",
    )
    table.wait_until_exists()

    yield table

    # Cleanup (optional — container dies anyway)
    table.delete()


@pytest.fixture(scope="session")
def redis_client():
    """Isolated Redis client with unique key prefix."""
    import redis
    session_id = uuid.uuid4().hex[:8]
    client = redis.Redis(host="localhost", port=6379, db=0)

    class PrefixedRedis:
        """Wraps Redis client to add session prefix to all keys."""
        def __init__(self, client, prefix):
            self._client = client
            self._prefix = f"test:{prefix}:"

        def get(self, key):
            return self._client.get(f"{self._prefix}{key}")

        def set(self, key, value, **kwargs):
            return self._client.set(f"{self._prefix}{key}", value, **kwargs)

        def setex(self, key, ttl, value):
            return self._client.setex(f"{self._prefix}{key}", ttl, value)

        def delete(self, *keys):
            return self._client.delete(*[f"{self._prefix}{k}" for k in keys])

    yield PrefixedRedis(client, session_id)

# .github/workflows/integration-tests.yml
name: Integration Tests

on: [pull_request]

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install -r requirements-test.txt
      - run: pytest tests/unit --tb=short -q  # 30 seconds

  integration-tests:
    runs-on: ubuntu-latest
    needs: unit-tests  # Only run if unit tests pass
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      # Start all test dependencies
      - name: Start test infrastructure
        run: |
          docker-compose -f docker-compose.test.yml up -d
          # Wait for all services to be healthy
          docker-compose -f docker-compose.test.yml exec -T \
            localstack awslocal dynamodb list-tables  # Verify DynamoDB is ready
          docker-compose -f docker-compose.test.yml exec -T \
            redis redis-cli ping                      # Verify Redis is ready

      - name: Run integration tests
        run: |
          pip install -r requirements-test.txt
          pytest tests/integration \
            --tb=short -q \
            --timeout=60 \
            -x  # Stop on first failure

      - name: Teardown
        if: always()
        run: docker-compose -f docker-compose.test.yml down -v

# docker-compose.test.yml — CI test infrastructure
version: "3.8"
services:
  localstack:
    image: localstack/localstack:3.0
    ports:
      - "4566:4566"
    environment:
      - SERVICES=dynamodb,s3,sqs,kinesis
      - DEFAULT_REGION=us-east-1
      - EAGER_SERVICE_LOADING=1   # Start all services immediately
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:4566/_localstack/health"]
      interval: 5s
      timeout: 3s
      retries: 10

  redis:
    image: redis:7.2-alpine
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

  opensearch:
    image: opensearchproject/opensearch:2.11.0
    ports:
      - "9200:9200"
    environment:
      - discovery.type=single-node
      - DISABLE_SECURITY_PLUGIN=true
      - OPENSEARCH_JAVA_OPTS=-Xms512m -Xmx512m
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9200/_cluster/health"]
      interval: 10s
      timeout: 5s
      retries: 10

  wiremock:
    image: wiremock/wiremock:3.3.1
    ports:
      - "8089:8080"
    volumes:
      - ./tests/wiremock:/home/wiremock  # Scenario JSON files
    command: ["--verbose"]

// tests/wiremock/mappings/circuit-breaker-test.json
// Simulates a slow-then-failing downstream service
{
  "mappings": [
    {
      "scenarioName": "slow-503-circuit-breaker",
      "requiredScenarioState": "Started",
      "newScenarioState": "Attempt-2",
      "request": {
        "method": "POST",
        "url": "/api/v1/inference"
      },
      "response": {
        "status": 503,
        "fixedDelayMilliseconds": 4800,
        "headers": { "Content-Type": "application/json" },
        "body": "{\"error\": \"Service Unavailable\"}"
      }
    },
    {
      "scenarioName": "slow-503-circuit-breaker",
      "requiredScenarioState": "Attempt-2",
      "newScenarioState": "Attempt-3",
      "request": {
        "method": "POST",
        "url": "/api/v1/inference"
      },
      "response": {
        "status": 503,
        "fixedDelayMilliseconds": 4800,
        "body": "{\"error\": \"Service Unavailable\"}"
      }
    },
    {
      "scenarioName": "slow-503-circuit-breaker",
      "requiredScenarioState": "Attempt-3",
      "newScenarioState": "Recovered",
      "request": {
        "method": "POST",
        "url": "/api/v1/inference"
      },
      "response": {
        "status": 200,
        "body": "{\"result\": \"recovered\"}"
      }
    }
  ]
}

# Test that caught the circuit breaker timing bug
import pytest
import time
import httpx

async def test_circuit_breaker_respects_per_attempt_timeout():
    """
    Bug this caught: Circuit breaker had 3 retries with NO per-attempt timeout.
    Downstream took 4.8s per 503 → total 14.4s before breaker opened.
    SLA was 10 seconds. This test enforces per-attempt timeout of 3 seconds.
    """
    start = time.monotonic()

    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.post(
            "http://localhost:8080/api/chat",
            json={"query": "test", "history": []},
        )

    elapsed = time.monotonic() - start

    # Circuit breaker should open within 10 seconds (our SLA)
    assert elapsed < 10.0, (
        f"Circuit breaker took {elapsed:.1f}s — exceeds 10s SLA. "
        "Check per-attempt timeout configuration."
    )

    # Should get a graceful degradation response, NOT a timeout error
    assert response.status_code in (200, 503)
    if response.status_code == 503:
        assert "capacity" in response.json().get("error", "")

╔══════════════════════════════════════════════════════════════════════╗
║                    CRITICAL POINTS — LLD-5                         ║
╠══════════════════════════════════════════════════════════════════════╣
║                                                                      ║
║  1. MOCKS TEST YOUR CODE. DOCKER TESTS THE INTERACTION.            ║
║     • Mock = your assumption about dependency behavior              ║
║     • Docker = dependency's ACTUAL behavior                        ║
║     • Mocks are blind to: ordering, TTL, serialization, latency    ║
║                                                                      ║
║  2. THE TEST TIER RULE                                             ║
║     • Unit (mocks): business logic → 30 seconds                   ║
║     • Docker integration: serialization/TTL/retry → 5 minutes      ║
║     • Staging: ML latency/cold-start → 10 minutes                 ║
║     • Place each test at the CHEAPEST tier that catches the bug    ║
║                                                                      ║
║  3. LOCALSTACK IS 90-95% COMPATIBLE, NOT 100%                      ║
║     • Covers: CRUD, basic queries, S3 operations                   ║
║     • Misses: throttling, IAM policy eval, eventual consistency    ║
║     • Use for functional correctness, not operational correctness  ║
║                                                                      ║
║  4. TEST ISOLATION = NAMESPACE PER TEST, NOT CONTAINER PER TEST    ║
║     • UUID-prefixed table names, key prefixes, DB numbers          ║
║     • Enables parallelism without data collision                   ║
║     • Container startup is expensive; namespace creation is free   ║
║                                                                      ║
║  5. WIREMOCK FOR FAILURE CHOREOGRAPHY                              ║
║     • Script exact failure sequences: 503 → 503 → 200             ║
║     • Add realistic delays (4.8s timeout, not instant)             ║
║     • Catches timing bugs that instant mocks ALWAYS miss           ║
║                                                                      ║
║  6. CI MUST BE FAST OR NOBODY RUNS IT                              ║
║     • Container reuse (start once, test many)                      ║
║     • Parallel test execution (pytest-xdist)                       ║
║     • Test categorization (fast on PR, full on merge)              ║
║     • Target: <5 minutes for PR feedback loop                      ║
║                                                                      ║
║  7. KEEP STAGING FOR WHAT DOCKER CAN'T FAKE                       ║
║     • Real SageMaker inference latency                             ║
║     • Real cold-start behavior                                     ║
║     • Real throttling and rate limiting                            ║
║     • Everything else → Docker containers in CI                    ║
║                                                                      ║
╚══════════════════════════════════════════════════════════════════════╝

flowchart LR
    Src["Reviewed source"] --> Lock["Lockfile + hashes"]
    Lock --> Build["Deterministic build"]
    Build --> Scan["SCA / CVE scan"]
    Scan --> SBOM["SBOM + dependency diff"]
    SBOM --> Sign["Signature + provenance attestation"]
    Sign --> ECR["Publish immutable image"]
    ECR --> Policy{"Promotion policy passed?"}
    Policy -->|No| Block["Block release"]
    Policy -->|Yes| Canary["Canary rollout"]
    Canary --> Healthy{"Healthy?"}
    Healthy -->|No| Rollback["Rollback to previous signed artifact"]
    Healthy -->|Yes| Prod["Promote to production"]

Supply Chain Attack Vector:

1. Attacker compromises a popular PyPI package (e.g., typosquatting)
2. Your CI pulls the latest version during `pip install`
3. The malicious package has no known CVE yet → scanner says CLEAN ✓
4. Image is pushed to ECR → scanner says CLEAN ✓
5. Image is deployed to production → scanner says CLEAN ✓
6. Malicious code executes → data exfiltration begins

What WOULD have caught this with a governed pipeline:

Step 2: Private mirror doesn't have the package → build FAILS ✗
   OR: Lockfile hash doesn't match → build FAILS ✗
   OR: SBOM diff shows unexpected new dependency → human reviews
   OR: Dependency diff flags the new package → promotion BLOCKED

Each layer catches what the previous one might miss.

Source ──► Build ──► Verify ──► Publish ──► Gate ──► Deploy

Source:
├── Code reviewed and merged to main
├── requirements.txt with exact versions AND hashes
│   Example: flask==3.0.0 --hash=sha256:abc123...
└── All deps available in private mirror

Build:
├── Deterministic: same inputs → same image bytes
├── pip install from private mirror ONLY (no pypi.org)
├── Multi-stage build: build deps never reach runtime image
└── Image tagged with git SHA (immutable)

Verify:
├── CVE scan (Trivy, Grype, or Snyk)
├── SBOM generation (Syft → CycloneDX format)
├── Dependency diff vs previous release
│   "These 3 packages changed versions: X 1.2→1.3, Y 2.0→2.1, Z 3.0 (NEW)"
├── Provenance attestation (Sigstore/cosign)
│   "Built by GitHub Actions run #4521 from commit abc123"
└── Digital signature (cosign sign)

Gate (all must pass — fail-closed):
├── ✓ No critical/high CVEs
├── ✓ SBOM present and valid
├── ✓ Signature verified
├── ✓ Provenance attestation present
├── ✓ Dependency diff reviewed (if new deps)
├── ✓ ML eval results present and passing
└── ✗ ANY missing evidence → release BLOCKED

Deploy:
├── Canary: 5% traffic for 15 minutes
├── Monitor error rate, latency, GPU metrics
├── If healthy → promote to 100%
└── If unhealthy → automatic rollback to PREVIOUS SIGNED DIGEST

# WITHOUT hashes (dangerous):
flask==3.0.0
# pip downloads flask 3.0.0 from PyPI
# But how do you know the bytes haven't been tampered with?
# Answer: You don't.

# WITH hashes (safe):
flask==3.0.0 \
    --hash=sha256:21c0527d5fce083e3dc580fa2b28db8c6f2add8a3e8... \
    --hash=sha256:7e8b2cdc7e7f5f2...
# pip downloads flask 3.0.0 AND verifies the SHA256 hash
# If the bytes don't match → install FAILS
# Tampered package → caught immediately

# Keyless signing workflow with cosign + Sigstore
# This runs in GitHub Actions CI after image build

# Step 1: Build and push image
IMAGE="123456789.dkr.ecr.us-east-1.amazonaws.com/chatbot-orchestrator"
TAG="${GITHUB_SHA}"
docker build -t "${IMAGE}:${TAG}" .
docker push "${IMAGE}:${TAG}"

# Step 2: Get the image digest (immutable reference)
DIGEST=$(docker inspect --format='{{index .RepoDigests 0}}' "${IMAGE}:${TAG}")
echo "Image digest: ${DIGEST}"

# Step 3: Sign with keyless cosign (uses OIDC identity from GitHub Actions)
cosign sign \
  --yes \
  --oidc-issuer=https://token.actions.githubusercontent.com \
  "${DIGEST}"

# Step 4: Attach SBOM as an attestation
syft "${DIGEST}" -o cyclonedx-json > sbom.json
cosign attest \
  --yes \
  --oidc-issuer=https://token.actions.githubusercontent.com \
  --predicate sbom.json \
  --type cyclonedx \
  "${DIGEST}"

# Step 5: Verify (this is what the promotion gate runs)
cosign verify \
  --certificate-oidc-issuer=https://token.actions.githubusercontent.com \
  --certificate-identity-regexp="github.com/our-org/our-repo" \
  "${DIGEST}"

echo "Signature verified. SBOM attached. Ready for promotion gate."

# Automated audit trail query — used by compliance dashboard
import boto3
import subprocess
import json

def audit_image(image_digest: str) -> dict:
    """
    Given a production image digest, reconstruct the full
    evidence chain from source to deployment.
    """
    audit = {"digest": image_digest, "evidence": {}}

    # 1. Verify signature and get build identity
    result = subprocess.run([
        "cosign", "verify",
        "--certificate-oidc-issuer=https://token.actions.githubusercontent.com",
        "--certificate-identity-regexp=github.com/our-org/.*",
        "--output-text",
        image_digest
    ], capture_output=True, text=True)
    sig_info = json.loads(result.stdout)
    audit["evidence"]["signature"] = {
        "verified": result.returncode == 0,
        "build_identity": sig_info[0]["optional"]["Subject"],
        "issuer": sig_info[0]["optional"]["Issuer"],
        "build_timestamp": sig_info[0]["optional"]["BuildSignerURI"],
    }

    # 2. Retrieve SBOM attestation
    result = subprocess.run([
        "cosign", "verify-attestation",
        "--type", "cyclonedx",
        "--certificate-oidc-issuer=https://token.actions.githubusercontent.com",
        image_digest
    ], capture_output=True, text=True)
    sbom = json.loads(result.stdout)
    audit["evidence"]["sbom"] = {
        "present": result.returncode == 0,
        "component_count": len(sbom.get("components", [])),
        "format": "CycloneDX",
    }

    # 3. ECR vulnerability scan results
    ecr = boto3.client("ecr")
    scan = ecr.describe_image_scan_findings(
        repositoryName="chatbot-orchestrator",
        imageId={"imageDigest": image_digest.split("@")[1]}
    )
    findings = scan["imageScanFindings"]["findingSeverityCounts"]
    audit["evidence"]["vulnerability_scan"] = {
        "critical": findings.get("CRITICAL", 0),
        "high": findings.get("HIGH", 0),
        "medium": findings.get("MEDIUM", 0),
        "scan_completed": scan["imageScanStatus"]["status"] == "COMPLETE",
    }

    # 4. Promotion gate record (from DynamoDB)
    dynamodb = boto3.resource("dynamodb")
    table = dynamodb.Table("deployment-promotions")
    gate_record = table.get_item(Key={"image_digest": image_digest})
    if "Item" in gate_record:
        audit["evidence"]["promotion_gate"] = {
            "all_gates_passed": gate_record["Item"]["all_passed"],
            "gates": gate_record["Item"]["gate_results"],
            "promoted_at": gate_record["Item"]["promoted_at"],
            "promoted_by": gate_record["Item"]["approved_by"],
        }

    return audit

02:00  PagerDuty alert fires from CVE feed integration
02:05  On-call opens SBOM dashboard, queries: "Which production images
       contain package X at version Y?"
       → Results in 8 seconds: image chatbot-inference:sha256:abc123
         deployed on production endpoint 'inference-prod'
02:10  Assess exploitability: Is the vulnerable code path reachable?
       → Check: Is this an HTTP parsing CVE? A compression library CVE?
       → If NOT reachable in our context → schedule patch for next release
       → If reachable → proceed to emergency patch
02:15  Branch from current release commit (from Rekor: commit sha def456)
       Update dependency version in requirements-lock.txt
       Regenerate lockfile hashes
02:20  CI builds new image → scan → SBOM → sign → all gates pass
02:25  Deploy to canary (5% traffic)
02:30  Canary healthy for 5 minutes → promote to 100%
02:35  Verify new image SBOM no longer contains vulnerable version
02:40  Update incident ticket. Stand down.
Total response: ~40 minutes from alert to patched production.
Without SBOM: Step 02:05 alone takes 2-4 hours (manual image inspection).

# .github/workflows/secure-release.yml
name: Secure Container Release

on:
  push:
    branches: [main]

permissions:
  id-token: write   # Needed for keyless cosign signing
  contents: read
  packages: write

jobs:
  build-and-verify:
    runs-on: ubuntu-latest
    outputs:
      image-digest: ${{ steps.push.outputs.digest }}

    steps:
      - uses: actions/checkout@v4

      # Build image
      - name: Build container image
        run: |
          docker build \
            --build-arg BUILD_DATE=$(date -u +%Y-%m-%dT%H:%M:%SZ) \
            --build-arg GIT_SHA=${{ github.sha }} \
            -t chatbot-orchestrator:${{ github.sha }} .

      # Push to ECR
      - name: Push to ECR
        id: push
        run: |
          IMAGE="123456789.dkr.ecr.us-east-1.amazonaws.com/chatbot-orchestrator"
          docker tag chatbot-orchestrator:${{ github.sha }} ${IMAGE}:${{ github.sha }}
          docker push ${IMAGE}:${{ github.sha }}
          DIGEST=$(docker inspect --format='{{index .RepoDigests 0}}' ${IMAGE}:${{ github.sha }})
          echo "digest=${DIGEST}" >> $GITHUB_OUTPUT

      # Vulnerability scan
      - name: Scan for CVEs
        uses: aquasecurity/trivy-action@master
        with:
          image-ref: ${{ steps.push.outputs.digest }}
          severity: CRITICAL,HIGH
          exit-code: 1  # Fail build on critical/high CVEs

      # Generate SBOM
      - name: Generate SBOM
        run: |
          syft ${{ steps.push.outputs.digest }} -o cyclonedx-json > sbom.json
          echo "Components found: $(jq '.components | length' sbom.json)"

      # Generate dependency diff vs previous release
      - name: Dependency diff
        run: |
          # Fetch previous release SBOM from S3
          aws s3 cp s3://release-artifacts/previous-sbom.json previous-sbom.json || true
          if [ -f previous-sbom.json ]; then
            python scripts/diff_sbom.py previous-sbom.json sbom.json > dep-diff.json
            echo "Dependency changes:"
            cat dep-diff.json
          fi

      # Sign image (keyless via Sigstore)
      - name: Sign image
        uses: sigstore/cosign-installer@v3
      - run: |
          cosign sign --yes ${{ steps.push.outputs.digest }}
          cosign attest --yes --predicate sbom.json --type cyclonedx \
            ${{ steps.push.outputs.digest }}

  # Promotion gate — fail-closed
  promotion-gate:
    needs: build-and-verify
    runs-on: ubuntu-latest
    steps:
      - name: Verify all evidence exists
        run: |
          DIGEST="${{ needs.build-and-verify.outputs.image-digest }}"

          echo "=== Checking signature ==="
          cosign verify \
            --certificate-oidc-issuer=https://token.actions.githubusercontent.com \
            --certificate-identity-regexp="github.com/our-org/.*" \
            "${DIGEST}" || { echo "SIGNATURE MISSING — BLOCKING RELEASE"; exit 1; }

          echo "=== Checking SBOM attestation ==="
          cosign verify-attestation --type cyclonedx \
            --certificate-oidc-issuer=https://token.actions.githubusercontent.com \
            "${DIGEST}" || { echo "SBOM MISSING — BLOCKING RELEASE"; exit 1; }

          echo "=== All gates passed ==="

  # Canary deployment
  canary-deploy:
    needs: [build-and-verify, promotion-gate]
    runs-on: ubuntu-latest
    steps:
      - name: Deploy canary (5% traffic)
        run: |
          aws ecs update-service \
            --cluster production \
            --service chatbot-orchestrator \
            --task-definition "chatbot-orchestrator:${{ github.sha }}" \
            --deployment-configuration "maximumPercent=200,minimumHealthyPercent=100"
          # CodeDeploy handles canary traffic splitting

╔══════════════════════════════════════════════════════════════════════╗
║                    CRITICAL POINTS — LLD-6                         ║
╠══════════════════════════════════════════════════════════════════════╣
║                                                                      ║
║  1. SCANNING ALONE IS NOT SECURITY                                 ║
║     • Scanning detects KNOWN CVEs only                             ║
║     • It does NOT prove: provenance, integrity, or reproducibility ║
║     • You need: scan + SBOM + signature + provenance + policy      ║
║                                                                      ║
║  2. FAIL-CLOSED PROMOTION = DEFAULT DENY                           ║
║     • Missing signature → BLOCKED                                  ║
║     • Missing SBOM → BLOCKED                                       ║
║     • Open critical CVE → BLOCKED                                  ║
║     • The absence of evidence IS the blocker                       ║
║                                                                      ║
║  3. DIGEST > TAG (always)                                          ║
║     • Tag = mutable label ("v2.1" can be overwritten)              ║
║     • Digest = content hash (sha256:abc123 is immutable)           ║
║     • All internal references use digest                           ║
║     • Tags are for human readability only                          ║
║                                                                      ║
║  4. LOCKFILE + HASHES = BYTE-LEVEL REPRODUCIBILITY                 ║
║     • Version pin: "flask==3.0.0" (locks version)                  ║
║     • Hash pin: "--hash=sha256:abc..." (locks exact bytes)         ║
║     • Tampered package → hash mismatch → build FAILS               ║
║     • Without hashes, version pin alone is insufficient            ║
║                                                                      ║
║  5. SBOM IS YOUR INCIDENT RESPONSE SUPERPOWER                     ║
║     • New CVE announced → query SBOM DB → impacted images in <10s  ║
║     • Without SBOM → manual inspection → hours of response time   ║
║     • SBOM format: CycloneDX or SPDX (both machine-readable)      ║
║                                                                      ║
║  6. KEYLESS SIGNING (Sigstore/cosign)                              ║
║     • No persistent private key to steal                           ║
║     • OIDC identity (CI system identity) = signing identity        ║
║     • Rekor transparency log = tamper-evident audit trail           ║
║     • Verifiable: "This was signed by GH Actions run #X on repo Y" ║
║                                                                      ║
║  7. ROLLBACK TO SIGNED DIGEST, NEVER TO TAG                       ║
║     • Promotion DB stores: digest + signature + gate evidence      ║
║     • Rollback query: "previous deployment's verified digest"      ║
║     • Verify signature on rollback target (hasn't been revoked)    ║
║     • Total rollback: <3 min Fargate, <5 min SageMaker             ║
║                                                                      ║
║  8. PRIVATE MIRROR = SUPPLY CHAIN FIREWALL                         ║
║     • All packages from approved mirror, never live PyPI           ║
║     • Approved packages list maintained in Git (PR-based process)  ║
║     • Prevents typosquatting, dependency confusion, hijacking      ║
║                                                                      ║
╚══════════════════════════════════════════════════════════════════════╝