Unit 7: Production Security Engineering

# Generate SBOM for your Python project
syft dir:. -o cyclonedx-json > sbom.json

# Example output:
{
  "specVersion": "1.4",
  "version": 1,
  "metadata": {
    "timestamp": "2026-03-05T10:30:00Z",
    "component": {
      "type": "application",
      "name": "agent-security-system",
      "version": "1.0.0"
    }
  },
  "components": [
    {
      "type": "library",
      "name": "anthropic",
      "version": "0.7.15",
      "purl": "pkg:pypi/anthropic@0.7.15",
      "hashes": [
        {
          "alg": "SHA-256",
          "content": "abc123..."
        }
      ]
    },
    {
      "type": "library",
      "name": "fastapi",
      "version": "0.104.1",
      "purl": "pkg:pypi/fastapi@0.104.1"
    }
  ]
}

# Generate signing keys
cosign generate-key-pair

# Sign a model file (e.g., model weights)
cosign sign-blob --key cosign.key model-weights.safetensors > model-weights.safetensors.sig

# Verify signature before loading
cosign verify-blob --key cosign.pub --signature model-weights.safetensors.sig model-weights.safetensors
# Output: Verified OK

# In Python, verify before loading:
import subprocess
import torch

def load_verified_model(model_path, pub_key):
    # Verify signature
    result = subprocess.run([
        'cosign', 'verify-blob',
        '--key', pub_key,
        '--signature', f'{model_path}.sig',
        model_path
    ], capture_output=True, text=True)

    if result.returncode != 0:
        raise SecurityError(f"Model signature verification failed: {result.stderr}")

    # Only load if verification succeeded
    return torch.load(model_path)

# In your project root, every commit:
pip install safety
safety check --json > vulnerability-report.json

# If vulnerabilities found with severity >= HIGH, CI/CD fails

I need a Python module that scans an AI project for supply chain vulnerabilities.

The module should:
1. Scan dependencies in requirements.txt using the 'safety' tool
2. Parse requirements.txt and extract direct dependencies with versions
3. Identify outdated packages (packages not updated in 6+ months)
4. Generate a JSON report with format:
   {
     "timestamp": "...",
     "findings": {
       "dependencies": {...},
       "vulnerabilities": {...},
       "outdated": [...]
     },
     "risk_score": "CRITICAL|HIGH|MEDIUM|LOW"
   }
5. Calculate risk score: 5+ vulns = CRITICAL, 2-4 = HIGH, 1 = MEDIUM, 0 = LOW

The main class should be called DependencyScanner with methods:
- scan_with_safety()
- parse_requirements()
- check_outdated_packages()
- generate_report()

Use subprocess to call external tools (safety, pip). Handle timeouts and errors gracefully.

Create a ModelProvenanceVerifier class that audits model artifacts in an AI project.

The class should:
1. Find all model files by glob patterns:
   - *.pt, *.pth (PyTorch)
   - *.h5 (Keras/TensorFlow)
   - *.onnx (ONNX)
   - *.safetensors (Hugging Face)
   - *.pkl (Pickle)
   Skip venv/ and __pycache__/ directories

2. Verify MODEL_CARD.md documentation:
   - Check file exists in model's directory
   - Verify it contains sections: "Model Details", "Intended Use", "Training Data", "Limitations", "Ethical Considerations"
   - Return: {has_card: bool, status: "OK|INCOMPLETE|CRITICAL", missing_sections: [...]}

3. Verify model signature:
   - Look for {model_path}.sig file containing expected SHA256 hash
   - Compute SHA256 of actual model file
   - Compare: if match, status="OK"; if no .sig file, status="NO_SIGNATURE"; if mismatch, status="VERIFICATION_FAILED"

4. Check model age:
   - Get file modification time
   - Calculate days since last update
   - If > 180 days: status="CRITICAL"
   - If > 90 days: status="WARNING"
   - Otherwise: status="OK"

Return dict format for each check with status and message fields.

Create a SBOMGenerator class that wraps the Syft tool to generate SBOMs.

The class should:
1. Have a generate_sbom(project_root, output_format="cyclonedx-json") method that:
   - Calls: syft dir:{project_root} -o {output_format}
   - Runs with 60 second timeout
   - Parses JSON output from stdout
   - Returns: {status: "success"|"failed", format: str, component_count: int, sbom: dict, error: str}
   - If syft is not installed, catch FileNotFoundError and return helpful error message

2. Have a save_sbom(sbom_data, output_path) method that:
   - Writes JSON to file (pretty-printed, 2-space indent)
   - Suitable for committing to git version control

Handle edge cases:
- Subprocess timeout (return {status: "timeout"...})
- JSON parse error (return {status: "parse_error"...})
- Permission errors (return {status: "permission_denied"...})

Return consistent dict format for all outcomes.

Create a SupplyChainRiskReport class that aggregates findings from DependencyScanner, ModelProvenanceVerifier, and SBOMGenerator.

The class should:
1. Constructor takes: scanner, verifier, sbom_gen
2. Main method: generate_report(project_root) that:
   - Calls scanner.generate_report() for dependency findings
   - Calls verifier.find_models() and for each model calls verify_model_card() and check_model_age()
   - Calls sbom_gen.generate_sbom(project_root)
   - Aggregates all findings into "findings" array with {category, status, details} format
   - Calculates overall risk_score: if any finding is CRITICAL → report is CRITICAL, else if any HIGH → HIGH, etc.
   - Generates recommendations based on critical findings

3. Method _generate_recommendations(report) that:
   - Iterates findings array
   - For each CRITICAL finding: adds "CRITICAL: {category} - Address immediately"
   - For each HIGH finding: adds "HIGH: {category} - Address before next release"
   - Returns list of recommendation strings

4. Support outputting to JSON (pretty-printed)

Return format:
{
  "executive_summary": "string",
  "risk_score": "CRITICAL|HIGH|MEDIUM|LOW",
  "findings": [...],
  "recommendations": [...]
}

Can Agent "AutoResponse" write to "incident_queue"?

Check attributes:
  - agent.role == "ResponseAutomation" ✓
  - request.time in 09:00-17:00 (business hours) ✓
  - request.resource.severity >= HIGH ✓
  - agent.last_activity < 5_minutes_ago ✓

Result: ALLOW

class CredentialRotation:
    """Automated rotation of non-human identities."""

    def rotate_api_key(self, agent_id: str) -> Dict:
        """Issue new API key, revoke old one."""
        # 1. Generate new key
        new_key = self.generate_api_key()

        # 2. Distribute to agent
        self.vault.set_secret(f"agent/{agent_id}/api_key", new_key)

        # 3. Mark old key for revocation (grace period)
        old_key_id = self.get_current_key_id(agent_id)
        self.revocation_queue.add(
            key_id=old_key_id,
            agent_id=agent_id,
            revoke_at=datetime.now() + timedelta(hours=1)  # Grace period
        )

        # 4. Log rotation
        self.audit_log.record({
            "event": "credential_rotated",
            "agent_id": agent_id,
            "new_key_id": new_key_id,
            "old_key_id": old_key_id,
            "timestamp": datetime.now().isoformat()
        })

        return {"status": "rotated", "effective_at": datetime.now()}

    def automatic_rotation_job(self):
        """Run every 24 hours: rotate credentials older than 90 days."""
        agents = self.get_all_agents()

        for agent in agents:
            cred_age = self.get_credential_age(agent.id)

            if cred_age > timedelta(days=90):
                self.rotate_api_key(agent.id)

class JITAccessManager:
    """Issue time-limited credentials on-demand."""

    def request_access(self, agent_id: str, resource: str,
                      duration_minutes: int = 15) -> Dict:
        """Agent requests temporary access to resource."""

        # 1. Check if agent is allowed to access this resource
        policy = self.load_policy(agent_id)

        if not policy.allows(resource, agent_id):
            self.audit_log.record({
                "event": "access_denied",
                "agent_id": agent_id,
                "resource": resource,
                "reason": "not_in_policy"
            })
            raise AccessDenied(f"Agent {agent_id} cannot access {resource}")

        # 2. Issue time-limited credential (token expires in duration_minutes)
        token = self.generate_token(
            agent_id=agent_id,
            resource=resource,
            expires_in=timedelta(minutes=duration_minutes)
        )

        # 3. Log the access request
        self.audit_log.record({
            "event": "jit_token_issued",
            "agent_id": agent_id,
            "resource": resource,
            "token_id": token.id,
            "expires_at": token.expires_at,
            "timestamp": datetime.now().isoformat()
        })

        return {
            "token": token.value,
            "expires_in_minutes": duration_minutes,
            "expires_at": token.expires_at.isoformat()
        }

{
  "event": "tool_call",
  "timestamp": "2026-03-05T10:30:00Z",
  "agent_id": "threat-analyzer-prod-v2",
  "action": "query_threat_intel_api",
  "resource": "threat_intel_api",
  "input": {
    "indicator": "192.0.2.1",
    "type": "ip"
  },
  "output": {
    "reputation_score": 85,
    "threat_level": "HIGH"
  },
  "duration_ms": 234,
  "status": "success",
  "authorization": {
    "policy_version": "v2.1.4",
    "attributes": {
      "agent.role": "ThreatAnalysis",
      "request.time": "2026-03-05T10:30:00Z",
      "request.severity": "prod"
    },
    "decision": "ALLOW"
  }
}

Create identity management classes for NHI governance:

1. IdentityTier enum with values: TIER_1="CRITICAL", TIER_2="HIGH", TIER_3="MEDIUM", TIER_4="LOW"

2. NonHumanIdentity dataclass with fields:
   - identity_id: str (unique identifier)
   - identity_type: str ("agent", "service_account", "api_key", "bot", etc.)
   - name: str (human-readable name)
   - tier: IdentityTier (risk classification)
   - created_at: datetime
   - last_rotated: datetime (when credential was last rotated)
   - expires_at: datetime (when credential expires)
   - enabled: bool = True
   - assigned_roles: list[str] = [] (e.g., ["ThreatAnalysis", "DataRead"])

   Methods:
   - is_expired() -> bool
   - days_until_rotation(rotation_days=90) -> int

3. IdentityRegistry class with methods:
   - __init__(): initialize empty registry and audit log
   - register_identity(identity: NonHumanIdentity): add new identity and log event
   - get_identity(identity_id: str) -> NonHumanIdentity
   - list_identities_by_tier(tier: IdentityTier) -> list[NonHumanIdentity]
   - get_expiring_credentials(days_until=30) -> list[NonHumanIdentity]
   - save_registry(filepath: str): persist to JSON with all identities

4. Example usage showing registration of 2-3 sample identities (threat analyzer agent, data pipeline service account)

Output should be JSON-serializable for compliance audits.

Create policy-as-code classes for NHI authorization:

1. Policy class with constructor taking identity_id and rules list.
   Rules are dicts with structure:
   {
     "effect": "ALLOW" or "DENY",
     "action": "call_tool:threat_intel" or "write" or "delete",
     "resource": "threat_intel_api" or "incident_reports" or "*",
     "conditions": {
       "time_window": "09:00-17:00" (optional),
       "max_rate": 1000 (optional)
     }
   }

2. Policy.can_perform_action(action: str, resource: str, context: dict) -> tuple[bool, str]:
   Logic:
   - Find all rules matching (action, resource)
   - If any rule has effect="DENY", return (False, "Explicit DENY")
   - If no rules match, return (False, "No matching policy")
   - For matching ALLOW rules, check conditions:
     - time_window: parse "HH:MM-HH:MM", get current hour, check if within window
     - max_rate: get context["current_rate"], check if < max_rate
   - If all conditions pass, return (True, "ALLOW")
   - If conditions fail, return (False, "reason")

3. PolicyStore class with methods:
   - set_policy(identity_id: str, policy: Policy)
   - get_policy(identity_id: str) -> Policy

4. Example: Create a threat analyzer policy with 3 rules:
   - ALLOW call_tool:threat_intel on threat_intel_api (no time window, max 5000/hr)
   - ALLOW write on incident_reports (max 100/hr)
   - DENY delete on * (all deletes forbidden)

Use datetime.now() to get current time for time window checks.

Create an AuthorizationEngine class that gates all agent actions:

Constructor takes:
- policy_store: PolicyStore instance
- audit_logger: AuditLogger instance (see Step 4)

Main method: authorize_action(agent_id, action, resource, context=None) -> dict
Logic:
1. Get policy for agent_id from policy_store
2. If no policy, return {allowed: False, decision: "DENY", reason: "No policy found"}
3. Get current rate via _get_current_rate(agent_id, resource)
4. Add current_rate to context dict
5. Call policy.can_perform_action(action, resource, context) to evaluate
6. Log authorization decision to audit_logger
7. If allowed, generate temporary token via _generate_temp_token()
8. Return {allowed: bool, decision: "ALLOW"|"DENY", reason: str, token: str|None}

Helper methods:
- _get_current_rate(agent_id, resource) -> int:
  Track calls per hour per agent/resource combo
  Use dict with key="agent:resource" and value={count: int, reset_at: datetime}
  Reset count if current time > reset_at
  Reset time should be 1 hour in future
  Return the current count

- _increment_rate(agent_id, resource):
  Increment count for agent:resource combo

- _generate_temp_token(agent_id, resource) -> str:
  Generate token as: f"{agent_id}:{resource}:{timestamp()}"
  Add timestamp so tokens are unique and can be validated against time

Include example usage showing agent authorization attempt.

Create an AuditLogger class for NHI audit trail and alerting:

Constructor takes log_file path (default: "nhi_audit.jsonl")

Methods:
1. log_authorization(event: dict): Write authorization event to log. Event should have timestamp, agent_id, action, resource, decision, reason.

2. log_access(agent_id, resource, action, result): Log access attempt (success or failure). Create event with timestamp, event_type="access", agent_id, action, resource, result dict. Write to log. Check for alerts.

3. log_credential_rotation(agent_id, old_key_id, new_key_id): Log credential rotation event with timestamp, event_type="credential_rotated", agent_id, key IDs.

4. _write_log(event): Append event as JSON to log file (JSONL format: one JSON per line). This is immutable—never modify old lines.

5. _check_for_alerts(event): Generate alerts for suspicious activity:
   - If event decision is "DENY": add {severity: "MEDIUM", message: "Unauthorized access attempt by {agent_id}"}
   - If action is "update_policy": add {severity: "HIGH", message: "Policy changed for {agent_id} at {timestamp}"}

6. query_audit_log(agent_id=None, event_type=None, start_time=None) -> list[dict]:
   Read log file line by line, parse JSON
   Filter by agent_id (if provided)
   Filter by event_type (if provided)
   Return list of matching events

Store alerts in self.alerts list.

Create an NHIGovernanceDashboard class:

Constructor takes:
- registry: IdentityRegistry
- audit_logger: AuditLogger

Methods:
1. generate_dashboard() -> dict: Return dict with:
   {
     "timestamp": "...",
     "summary": {
       "total_identities": int,
       "by_tier": {TIER_1: count, TIER_2: count, ...},
       "expiring_soon": int,
       "disabled": int
     },
     "alerts": {
       "high_severity": int,
       "medium_severity": int,
       "recent_alerts": list of last 5 alerts
     },
     "rotation_schedule": list of 5 next-to-expire credentials,
     "compliance": {
       "identities_with_time_bounds": int or %,
       "least_privilege_adoption": float %,
       "audit_log_entries": int
     }
   }

2. _count_by_tier() -> dict: Count identities in each IdentityTier

3. _count_disabled() -> int: Count disabled identities

4. _count_with_expiry() -> int: Count identities with expires_at field

5. _least_privilege_adoption() -> float:
   Least privilege = identity has <= 2 roles
   Return: (count of least privilege / total identities) * 100

6. _next_rotations() -> list[dict]:
   Get credentials expiring within 30 days
   Sort by expiration date
   Return first 5 with {identity_id, name, expires_at, days_remaining}

7. _audit_log_count() -> int: Count lines in audit log file

8. print_dashboard(): Print human-readable version of dashboard

Include example showing dashboard being generated and printed.

from opentelemetry import trace, metrics
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.exporter.prometheus import PrometheusMetricReader
import time

# Setup Jaeger exporter for traces
jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)

trace.set_tracer_provider(
    TracerProvider(
        resource_attributes={
            "service.name": "agent-security-system",
            "service.version": "1.0.0"
        }
    )
)
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(jaeger_exporter)
)

tracer = trace.get_tracer(__name__)

# Setup Prometheus metrics
metrics.set_meter_provider(MeterProvider(
    metric_readers=[PrometheusMetricReader()]
))
meter = metrics.get_meter(__name__)

# Define metrics
agent_duration = meter.create_histogram(
    name="agent_execution_duration_ms",
    description="Duration of agent execution",
    unit="ms"
)

agent_errors = meter.create_counter(
    name="agent_errors_total",
    description="Total agent errors",
    unit="1"
)

tokens_per_task = meter.create_histogram(
    name="tokens_per_task",
    description="Token usage per task",
    unit="tokens"
)

cost_per_task = meter.create_histogram(
    name="cost_per_task_usd",
    description="Cost per task in USD",
    unit="USD"
)

# Instrument agent execution
class InstrumentedAgent:
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.tracer = tracer
        self.meter = meter

    def execute_task(self, task: str) -> str:
        """Execute task with full observability."""
        with self.tracer.start_as_current_span("agent_execute") as span:
            span.set_attribute("agent.id", self.agent_id)
            span.set_attribute("task", task)

            start_time = time.time()
            error_occurred = False
            tokens_used = 0

            try:
                # Call tool 1
                with self.tracer.start_as_current_span("tool_call") as tool_span:
                    tool_span.set_attribute("tool.name", "lookup_threat_intel")
                    tool_span.set_attribute("tool.input", "192.0.2.1")

                    result1 = self._lookup_threat_intel("192.0.2.1")
                    tokens_used += 150  # Assume tool used 150 tokens

                    tool_span.set_attribute("tool.result", result1)

                # Call tool 2
                with self.tracer.start_as_current_span("tool_call") as tool_span:
                    tool_span.set_attribute("tool.name", "analyze_threat")
                    tool_span.set_attribute("tool.input", result1)

                    result2 = self._analyze_threat(result1)
                    tokens_used += 200  # Assume tool used 200 tokens

                    tool_span.set_attribute("tool.result", result2)

                return result2

            except Exception as e:
                error_occurred = True
                agent_errors.add(1, {"agent": self.agent_id})
                span.record_exception(e)
                span.set_attribute("error", True)
                raise

            finally:
                # Record metrics
                duration_ms = (time.time() - start_time) * 1000
                agent_duration.record(duration_ms, {"agent": self.agent_id})

                tokens_per_task.record(tokens_used, {"agent": self.agent_id})

                # Calculate cost (example: $0.0001 per 1K tokens)
                cost = (tokens_used / 1000) * 0.0001
                cost_per_task.record(cost, {"agent": self.agent_id})

    def _lookup_threat_intel(self, indicator: str) -> str:
        """Simulate tool call."""
        time.sleep(0.1)
        return f"Reputation score for {indicator}: HIGH"

    def _analyze_threat(self, intel: str) -> str:
        """Simulate tool call."""
        time.sleep(0.05)
        return f"Analysis: {intel} - Recommend isolation"

# Usage
agent = InstrumentedAgent("threat-analyzer-prod")
result = agent.execute_task("analyze IP 192.0.2.1")

class TokenTracker:
    """Track token usage by agent, task, user."""

    def __init__(self):
        self.token_log = []  # List of token events

    def log_token_usage(self, agent_id: str, task_id: str,
                       input_tokens: int, output_tokens: int,
                       model: str):
        """Log tokens used for a single API call."""
        event = {
            "timestamp": datetime.now().isoformat(),
            "agent_id": agent_id,
            "task_id": task_id,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total_tokens": input_tokens + output_tokens,
            "model": model,
            "cost_usd": self._calculate_cost(model, input_tokens, output_tokens)
        }
        self.token_log.append(event)
        return event

    def _calculate_cost(self, model: str, input_tokens: int,
                       output_tokens: int) -> float:
        """Calculate cost based on model pricing."""
        # Anthropic Claude 4.5 family pricing (as of March 2026)
        pricing = {
            "claude-sonnet-4-6": {
                "input": 0.003 / 1_000_000,      # $3 per 1M tokens
                "output": 0.015 / 1_000_000      # $15 per 1M tokens
            },
            "claude-opus-4-6": {
                "input": 0.015 / 1_000_000,
                "output": 0.075 / 1_000_000
            }
        }

        if model not in pricing:
            return 0.0

        prices = pricing[model]
        return (input_tokens * prices["input"]) + (output_tokens * prices["output"])

    def get_cost_summary(self, groupby: str = "agent") -> Dict[str, float]:
        """Aggregate costs by agent, task, or time."""
        summary = {}

        for event in self.token_log:
            key = event[groupby]
            summary[key] = summary.get(key, 0) + event["cost_usd"]

        return summary

    def detect_cost_anomaly(self, threshold_percentile: float = 0.95) -> List[Dict]:
        """Detect unusually expensive tasks."""
        if not self.token_log:
            return []

        import statistics
        costs = [e["cost_usd"] for e in self.token_log]

        threshold = statistics.quantiles(costs, n=100)[int(threshold_percentile * 100)]

        anomalies = [
            e for e in self.token_log
            if e["cost_usd"] > threshold
        ]

        return anomalies

# Example usage
tracker = TokenTracker()

# Agent calls Claude API
tracker.log_token_usage(
    agent_id="threat-analyzer-prod",
    task_id="incident-2026-03-05-001",
    input_tokens=500,
    output_tokens=1500,
    model="claude-sonnet-4-6"
)

# Another call
tracker.log_token_usage(
    agent_id="threat-analyzer-prod",
    task_id="incident-2026-03-05-002",
    input_tokens=200,
    output_tokens=800,
    model="claude-sonnet-4-6"
)

# Get cost summary
print(tracker.get_cost_summary(groupby="agent"))
# Output: {"threat-analyzer-prod": 0.000435}

# Detect anomalies
anomalies = tracker.detect_cost_anomaly()
print(f"Found {len(anomalies)} anomalous tasks")

0.95^5 = 0.77 (77% accuracy)

class ErrorCompoundingAnalyzer:
    """Analyze error propagation through agent chain."""

    def __init__(self):
        self.agent_accuracies = {}  # agent -> accuracy %
        self.failure_cases = []

    def add_agent_accuracy(self, agent_id: str, accuracy: float):
        """Record agent accuracy (0-100%)."""
        self.agent_accuracies[agent_id] = accuracy / 100.0

    def calculate_chain_accuracy(self, agent_chain: List[str]) -> float:
        """Calculate end-to-end accuracy."""
        product = 1.0
        for agent_id in agent_chain:
            if agent_id not in self.agent_accuracies:
                raise ValueError(f"No accuracy data for {agent_id}")
            product *= self.agent_accuracies[agent_id]
        return product * 100

    def identify_bottleneck(self, agent_chain: List[str]) -> Dict:
        """Find weakest agent in chain."""
        accuracies = [
            (agent_id, self.agent_accuracies[agent_id] * 100)
            for agent_id in agent_chain
        ]

        bottleneck_agent, bottleneck_accuracy = min(accuracies, key=lambda x: x[1])

        return {
            "bottleneck_agent": bottleneck_agent,
            "current_accuracy": bottleneck_accuracy,
            "impact": f"Improving to 99% would increase chain accuracy by {self._calculate_improvement(agent_chain, bottleneck_agent)}%"
        }

    def _calculate_improvement(self, agent_chain: List[str], target_agent: str) -> float:
        """Calculate impact of improving target agent to 99%."""
        current = self.calculate_chain_accuracy(agent_chain)

        # Modify accuracy
        original = self.agent_accuracies[target_agent]
        self.agent_accuracies[target_agent] = 0.99
        improved = self.calculate_chain_accuracy(agent_chain)
        self.agent_accuracies[target_agent] = original

        return improved - current

# Example
analyzer = ErrorCompoundingAnalyzer()
analyzer.add_agent_accuracy("threat-analyzer", 95)
analyzer.add_agent_accuracy("response-recommender", 90)
analyzer.add_agent_accuracy("risk-scorer", 94)

chain = ["threat-analyzer", "risk-scorer", "response-recommender"]

end_to_end = analyzer.calculate_chain_accuracy(chain)
print(f"End-to-end accuracy: {end_to_end:.1f}%")  # ~79%

bottleneck = analyzer.identify_bottleneck(chain)
print(f"Bottleneck: {bottleneck['bottleneck_agent']} ({bottleneck['current_accuracy']:.0f}%)")
print(bottleneck['impact'])

from enum import Enum

class EscalationLevel(Enum):
    RETRY = "retry"           # Try again, same agent
    FALLBACK = "fallback"      # Use alternate agent/rule
    DEGRADE = "degrade"        # Reduce functionality
    ESCALATE = "escalate"      # Human takes over

class ResilientAgent:
    """Agent with error recovery and escalation."""

    def __init__(self, agent_id: str, fallback_agent: str = None):
        self.agent_id = agent_id
        self.fallback_agent = fallback_agent
        self.max_retries = 3

    def execute_with_resilience(self, task: str, context: Dict) -> Dict:
        """Execute task with retry, fallback, degrade, escalate."""

        for attempt in range(self.max_retries):
            try:
                result = self._execute_task(task, context)
                return {"status": "success", "result": result}

            except TaskFailure as e:
                # Attempt 1-2: retry
                if attempt < self.max_retries - 1:
                    continue

                # Final attempt failed: fallback or escalate
                if self.fallback_agent:
                    try:
                        result = self._delegate_to_fallback(task, context)
                        return {
                            "status": "degraded",
                            "message": "Using fallback agent",
                            "result": result
                        }
                    except Exception:
                        pass

                # Fallback also failed: escalate to human
                return {
                    "status": "escalated",
                    "message": f"Agent and fallback both failed: {e}",
                    "escalation_level": EscalationLevel.ESCALATE.value,
                    "context": context
                }

    def _execute_task(self, task: str, context: Dict) -> str:
        """Execute task (may raise exception)."""
        # Simulate task execution
        raise TaskFailure("API timeout")

    def _delegate_to_fallback(self, task: str, context: Dict) -> str:
        """Use fallback agent/rule."""
        if self.fallback_agent == "pattern_match":
            return "Using pattern matching rules (degraded mode)"
        return ""

class TaskFailure(Exception):
    pass

# Example escalation workflow
agent = ResilientAgent(
    agent_id="threat-analyzer",
    fallback_agent="pattern_match"
)

result = agent.execute_with_resilience(
    task="analyze IP 192.0.2.1",
    context={"severity": "HIGH"}
)

if result["status"] == "escalated":
    print(f"ESCALATE TO HUMAN: {result['message']}")
    # Queue for human review

Create OpenTelemetry instrumentation for an agent system:

1. Setup (boilerplate):
   - Create JaegerExporter pointing to localhost:6831
   - Create TracerProvider and add BatchSpanProcessor
   - Create PrometheusMetricReader and MeterProvider
   - Get tracer = trace.get_tracer("agent-system")
   - Get meter = metrics.get_meter("agent-system")

2. Define metrics:
   - execution_duration: histogram in milliseconds
   - tool_call_latency: histogram in milliseconds
   - error_counter: counter for total errors (attribute: error_type)
   - tokens_counter: counter for total tokens (attribute: agent)

3. StructuredLogger class:
   - log(level, event, **kwargs): Create JSON dict with timestamp, level, event, all kwargs
   - Append to self.logs list
   - Print as JSON (could go to stdout or logging system)

4. ObservableAgent class:
   Constructor: agent_id, role, logger

   execute(task, input_data) -> dict:
   - Start span "agent.execute" with attributes: agent.id, agent.role, task
   - Log "agent.started"
   - Try:
     - For each tool (tool_1, tool_2):
       - Call _call_tool() and get tokens
     - Record execution_duration metric with agent and role tags
     - Record tokens_counter with agent tag
     - Log "agent.completed" with duration_ms and tokens_used
     - Return {status: "success", result: "..."}
   - Except:
     - Record error_counter
     - Record exception in span
     - Log "agent.failed"
     - Re-raise

   _call_tool(tool_name) -> int:
   - Start span "tool.call" with attribute tool.name
   - Simulate work (time.sleep(0.05))
   - Record tool_call_latency
   - Return token count (e.g., 200)

Include example showing agent execution with tracing.

Create a CostDashboard class for cost tracking and forecasting:

Constructor takes: token_tracker (TokenTracker instance)

Methods:
1. generate_dashboard() -> dict:
   - Get cost_by_agent = token_tracker.get_cost_summary(groupby="agent")
   - Calculate total_cost = sum of all costs
   - Calculate current_hour_cost = sum of costs for events where timestamp.hour == now.hour
   - Calculate daily_forecast = current_hour_cost * 24
   - Get anomalies = token_tracker.detect_cost_anomaly(threshold_percentile=0.95)
   - Return:
     {
       "timestamp": datetime.now().isoformat(),
       "cost": {
         "total_today": float,
         "daily_forecast": float,
         "by_agent": dict
       },
       "anomalies": {
         "count": int,
         "top_5": list of top 5 most expensive anomalies
       },
       "alerts": list of alert strings
     }

2. _generate_cost_alerts(total, forecast) -> list[str]:
   - Define daily_budget = 100.0 (example)
   - If forecast > budget: add alert "Forecast exceeds budget: ${forecast:.2f} vs ${budget:.2f}"
   - Return alerts list

3. print_dashboard():
   - Generate dashboard data
   - Print in human-readable format with sections:
     [COST SUMMARY]: total_today, daily_forecast
     [BY AGENT]: agents sorted by cost descending
     [ANOMALIES]: top 5 expensive anomalies
     [ALERTS]: any cost warnings

The TokenTracker.get_cost_summary(groupby="agent") returns dict like {agent_id: total_cost_usd, ...}
The anomaly detection finds tasks with costs > 95th percentile threshold.

Create a HealthDashboard class for system reliability monitoring:

Constructor takes:
- audit_logger: AuditLogger instance
- error_analyzer: ErrorCompoundingAnalyzer instance

Methods:
1. generate_dashboard() -> dict:
   - audit_entries = audit_logger.query_audit_log()
   - total_requests = len(audit_entries)
   - error_requests = count of entries where result.decision == "DENY"
   - success_rate = (1 - error_requests / max(total_requests, 1)) * 100
   - sample_chain = ["threat-analyzer", "risk-scorer", "response-recommender"]
   - chain_accuracy = error_analyzer.calculate_chain_accuracy(sample_chain)
   - escalations = count of entries where event_type == "escalation"
   - escalation_rate = (escalations / max(total_requests, 1)) * 100

   Return:
   {
     "timestamp": datetime.now().isoformat(),
     "system": {
       "total_requests": int,
       "success_rate_percent": float,
       "error_count": int
     },
     "quality": {
       "end_to_end_accuracy_percent": float,
       "human_escalations": int,
       "escalation_rate_percent": float
     },
     "alerts": list of alert strings
   }

2. _generate_health_alerts(success_rate, accuracy) -> list[str]:
   - If success_rate < 95: add "Success rate below SLO: {success_rate:.1f}%"
   - If accuracy < 80: add "End-to-end accuracy degraded: {accuracy:.1f}%"
   - Return alerts

3. print_dashboard():
   - Generate dashboard
   - Print human-readable format with sections:
     [PERFORMANCE]: total_requests, success_rate_percent
     [QUALITY]: end_to_end_accuracy_percent, human_escalations
     [ALERTS]: any warnings

SLO values: 95% success rate, 80% accuracy. Adjust if needed.

import anthropic
import os
import time
import logging

# Pattern 1: Key management via GitHub Secrets / environment variable
# In GitHub Actions: secrets.ANTHROPIC_API_KEY is injected as an env var.
# In a container: pass -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY at run time.
# Never hardcode the key or bake it into the image.
def get_claude_client():
    api_key = os.environ["ANTHROPIC_API_KEY"]  # raises KeyError if missing — fail fast
    return anthropic.Anthropic(api_key=api_key)

# Pattern 2: Retry with exponential backoff for rate limit errors
def call_claude_with_retry(client, messages, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=4096,
                messages=messages
            )
        except anthropic.RateLimitError:
            if attempt == max_retries - 1:
                raise
            wait = (2 ** attempt) * 1  # 1s, 2s, 4s
            logging.warning(f"Rate limited — retrying in {wait}s (attempt {attempt+1})")
            time.sleep(wait)

# Pattern 3: Cost monitoring — track token usage per call
def call_with_cost_tracking(client, messages, agent_id):
    response = call_claude_with_retry(client, messages)
    usage = response.usage
    # Emit as OTel metric — your OTel collector receives the spans
    logging.info("claude_api_call", extra={
        "agent_id": agent_id,
        "input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
        "model": "claude-sonnet-4-6"
    })
    return response
# Output filtering (Layer 2 enforcement) is applied in agent code, not at
# the API layer. Use NeMo Guardrails or a PostToolUse hook to check
# responses for PII, credential patterns, and off-topic content.

CloudFront / API Gateway (Authentication, Rate Limiting)
           ↓
    FastAPI Backend Service (Container: GHCR image, any Docker runtime)
           ↓
    ┌──────────────────────────────────────────────────┐
    │                                                  │
    ├── Business Logic (Threat Analysis, Scanning)    │
    ├── Database Layer (RDS, DynamoDB)               │
    ├── Cache Layer (ElastiCache Redis)               │
    └──────────────────────────────────────────────────┘
           ↑
    ┌──────────────┬─────────────────┬──────────────┐
    │              │                 │              │
    ├─ MCP Server  ├─ Web Dashboard  ├─ CLI Tools   │
    │ (Agent)      │ (Web UI)        │ (Ops)        │
    └──────────────┴─────────────────┴──────────────┘

from fastapi import FastAPI, Depends, HTTPException, RateLimiter
from fastapi.security import HTTPBearer
import structlog

app = FastAPI(title="Threat Analysis Service")
security = HTTPBearer()
logger = structlog.get_logger()

# Rate limiter: agents can't hammer the API
limiter = RateLimiter(calls=100, period=60)

@app.get("/api/v1/threat-intel/{ip_address}")
@limiter
async def get_threat_intel(
    ip_address: str,
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    """Threat intelligence lookup—callable by agents, dashboards, and CI/CD."""
    # Authentication & authorization
    agent_id = verify_jwt(credentials.credentials)
    if not authz.can_access(agent_id, "threat_intel"):
        raise HTTPException(status_code=403, detail="Access denied")

    # Validate input
    if not is_valid_ip(ip_address):
        raise HTTPException(status_code=400, detail="Invalid IP address")

    # Core business logic (decoupled from MCP)
    result = threat_db.lookup(ip_address)

    # Audit log
    logger.info("threat_intel_queried", ip=ip_address, agent_id=agent_id)

    return result

@app.post("/api/v1/scan")
@limiter
async def start_vulnerability_scan(
    target: ScanRequest,
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    """Start a security scan—MCP server calls this endpoint."""
    agent_id = verify_jwt(credentials.credentials)

    # Queue the scan for async processing
    scan_job = scanner.enqueue(
        target=target.ip_or_domain,
        requester=agent_id
    )

    logger.info("scan_initiated", target=target.ip_or_domain, job_id=scan_job.id)
    return {"job_id": scan_job.id, "status": "queued"}

@app.get("/health")
async def health_check():
    """Kubernetes uses this for liveness/readiness probes."""
    return {"status": "healthy"}

class ThreatAnalysisMCPServer:
    def __init__(self, api_base_url, agent_id, api_key):
        self.api_base_url = api_base_url
        self.agent_id = agent_id
        self.api_key = api_key

    def query_threat_intel(self, ip_address):
        """MCP tool wrapper—delegates to REST API."""
        response = requests.get(
            f"{self.api_base_url}/api/v1/threat-intel/{ip_address}",
            headers={"Authorization": f"Bearer {self.api_key}"}
        )
        response.raise_for_status()
        return response.json()

    def start_scan(self, target_ip):
        """MCP tool wrapper—delegates to REST API."""
        response = requests.post(
            f"{self.api_base_url}/api/v1/scan",
            json={"ip_or_domain": target_ip},
            headers={"Authorization": f"Bearer {self.api_key}"}
        )
        response.raise_for_status()
        return response.json()

# Build, tag, and push to GitHub Container Registry
# Run these from your terminal (or in the CI pipeline)

# Step 1: Authenticate to GHCR using a GitHub Personal Access Token (PAT)
#   Set GITHUB_TOKEN in your environment: export GITHUB_TOKEN=ghp_...
Login:
  command: echo $GITHUB_TOKEN | docker login ghcr.io -u YOUR_GITHUB_USERNAME --password-stdin

# Step 2: Build and tag the image
Build:
  command: docker build -t ghcr.io/YOUR_GITHUB_USERNAME/soc-agent:latest .

# Step 3: Scan for vulnerabilities before pushing
Scan:
  command: trivy image --severity CRITICAL ghcr.io/YOUR_GITHUB_USERNAME/soc-agent:latest

# Step 4: Push to GHCR (blocks if Trivy found CRITICAL CVEs)
Push:
  command: docker push ghcr.io/YOUR_GITHUB_USERNAME/soc-agent:latest

# Step 5: Run anywhere Docker is available
#   ANTHROPIC_API_KEY is injected at run time — never baked into the image
Run:
  command: docker run -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY ghcr.io/YOUR_GITHUB_USERNAME/soc-agent:latest

# GitHub Actions pipeline — security gates at every stage
DeploymentPipeline:
  stages:
    - test: Run pytest, bandit (security linting), and dependency checks
    - build: Build and scan Docker image with Trivy
    - push: Push to GHCR with vulnerability check gate (blocks on CRITICAL)
    - deploy_staging: Pull from GHCR and run in staging environment
    - approval: Human approval required before prod (GitHub Environments)
    - deploy_prod: Pull latest image tag from GHCR and redeploy
    - observe: OTel spans + container stdout/stderr for error rate and latency

# Multi-stage build for minimal image size
FROM python:3.11-slim as builder

# Install build dependencies
RUN apt-get update && apt-get install -y \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

WORKDIR /tmp
COPY requirements.txt .

# Create wheels
RUN pip install --no-cache-dir --user --wheel --no-deps --wheel-dir /tmp/wheels \
    -r requirements.txt

# Final stage
FROM python:3.11-slim

# Security: create non-root user
RUN groupadd -r agent && useradd -r -g agent agent

WORKDIR /app

# Copy only wheels from builder
COPY --from=builder /tmp/wheels /tmp/wheels
COPY --from=builder /tmp/requirements.txt .

# Install dependencies
RUN pip install --no-cache-dir --user --no-index --find-links /tmp/wheels \
    -r requirements.txt && \
    rm -rf /tmp/wheels

# Copy application
COPY --chown=agent:agent . .

# Security: run as non-root
USER agent

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

# Metadata
LABEL org.opencontainers.image.vendor="SecurityOrg" \
      org.opencontainers.image.version="1.0.0" \
      org.opencontainers.image.title="Agentic Security System"

EXPOSE 8000

ENTRYPOINT ["python", "-m", "agent_system.main"]
CMD ["--config", "/etc/agent/config.yaml"]

# GitHub Actions example
name: Deploy Agentic System

on:
  push:
    branches: [main, staging]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest safety snyk

      - name: Run unit tests
        run: pytest tests/unit/ -v

      - name: Run integration tests
        run: pytest tests/integration/ -v

      - name: Security scanning - Dependency check
        run: |
          safety check --json > safety-report.json
          # Fail if critical vulnerabilities
          if grep -q '"severity": "CRITICAL"' safety-report.json; then
            echo "Critical vulnerabilities found!"
            exit 1
          fi

      - name: Security scanning - SAST
        run: |
          pip install bandit
          bandit -r agent_system/ -f json -o bandit-report.json

      - name: SBOM generation
        run: |
          pip install syft
          syft dir:. -o cyclonedx-json > sbom.json

  build:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Docker buildx
        uses: docker/setup-buildx-action@v3

      - name: Login to Docker registry
        uses: docker/login-action@v3
        with:
          username: ${{ secrets.DOCKER_USERNAME }}
          password: ${{ secrets.DOCKER_PASSWORD }}

      - name: Build Docker image
        run: |
          docker build \
            -t agent-system:${{ github.sha }} \
            -t agent-system:latest \
            .

      - name: Scan Docker image for vulnerabilities
        run: |
          pip install trivy
          trivy image --severity CRITICAL agent-system:${{ github.sha }}

  deploy-staging:
    needs: build
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/staging'
    steps:
      - name: Deploy to staging
        run: |
          # Deploy to staging environment
          kubectl set image deployment/agent-system \
            agent=agent-system:${{ github.sha }} \
            --record

  deploy-canary:
    needs: build
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    steps:
      - name: Deploy canary (10% traffic)
        run: |
          # Istio/Flagger canary deployment
          kubectl set image deployment/agent-system-canary \
            agent=agent-system:${{ github.sha }}

  deploy-production:
    needs: deploy-canary
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    environment:
      name: production
      url: https://agent-system.prod.example.com
    steps:
      - name: Manual approval required
        run: echo "Waiting for approval..."

      - name: Deploy to production (100% traffic)
        run: |
          kubectl set image deployment/agent-system \
            agent=agent-system:${{ github.sha }} \
            --record

      - name: Verify deployment
        run: |
          kubectl rollout status deployment/agent-system -n production

# Istio VirtualService for canary
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: agent-system
spec:
  hosts:
  - agent-system.example.com
  http:
  - match:
    - uri:
        prefix: /
    route:
    - destination:
        host: agent-system-stable
        port:
          number: 8000
      weight: 90
    - destination:
        host: agent-system-canary
        port:
          number: 8000
      weight: 10

# Step 1: Deploy to green
kubectl apply -f agent-system-green.yaml

# Step 2: Run smoke tests against green
./smoke_tests.sh green.agent-system.local

# Step 3: Switch traffic (via load balancer)
kubectl patch service agent-system -p '{"spec":{"selector":{"version":"green"}}}'

# Step 4: Keep blue for quick rollback
# If issues detected, switch back to blue

class ShadowDeployment:
    """Compare new vs. old agent output without affecting users."""

    def process_request(self, request):
        # Primary (old) version
        primary_response = self.old_agent.process(request)

        # Shadow (new) version in parallel
        shadow_response = self.new_agent.process(request)

        # Compare outputs
        divergence = self._compare_outputs(primary_response, shadow_response)

        if divergence > THRESHOLD:
            self.logger.warn(f"Shadow divergence detected: {divergence}")

        # Return primary response (shadow doesn't affect users)
        return primary_response

    def _compare_outputs(self, primary, shadow):
        """Measure difference between responses."""
        # Example: token count difference
        if "tokens" in primary and "tokens" in shadow:
            return abs(primary["tokens"] - shadow["tokens"])
        return 0

class PreToolUseGate:
    """Intercept and approve/deny tool calls before execution."""

    def evaluate_call(self, agent_id, tool_name, tool_args):
        """
        Returns: APPROVE, DENY, or REQUIRE_APPROVAL
        """
        # Explicit blocklist at the MCP config level
        if tool_name in self.disallowed_commands:
            return DENY, "Tool blocked by policy"

        # Tool-specific guardrails
        if tool_name == "bash_execute":
            for blocked_cmd in self.disallowed_bash_commands:
                if blocked_cmd in tool_args["command"]:
                    return DENY, f"Command contains blocked pattern: {blocked_cmd}"

        # Data access guardrails
        if tool_name == "read_file":
            path = tool_args["file_path"]
            if not self._is_authorized_path(agent_id, path):
                return DENY, f"Agent {agent_id} not authorized to read {path}"

        # Unknown tools default to requiring human approval
        if tool_name not in self.known_tools:
            return REQUIRE_APPROVAL, f"Unknown tool: {tool_name}"

        return APPROVE, None

{
  "tools": [
    {
      "name": "bash_execute",
      "enabled": true,
      "disallowedBashCommands": [
        "rm -rf /",
        "sudo",
        "dd if=/dev/zero",
        ":(){ :|:& };:",
        "cat /etc/shadow",
        "curl http://attacker.com"
      ]
    }
  ]
}

class AgentRoleModel:
    """Agents have roles that constrain their permissions."""

    def __init__(self):
        self.roles = {
            "EXECUTOR": {
                # Can execute approved tasks
                "can_execute": True,
                "can_approve": False,  # Critical: agents cannot self-approve
                "tools": ["bash_execute", "api_call", "write_file"]
            },
            "REVIEWER": {
                # Can review other agents' outputs
                "can_execute": False,
                "can_approve": True,
                "tools": ["read_file", "create_report"]
            },
            "OPERATOR": {
                # Humans manage the system
                "can_execute": False,
                "can_approve": True,
                "tools": ["all"]
            }
        }

    def check_permission(self, agent_id, action):
        role = self.get_agent_role(agent_id)
        return action in self.roles[role].get(f"can_{action}", False)

class ContextDriftDetector:
    """Detect unauthorized changes to governance context."""

    def __init__(self):
        self.baseline_signatures = {
            "CLAUDE.md": hash_file("CLAUDE.md"),
            "system_prompt": hash_content(SYSTEM_PROMPT),
            ".mcp.json": hash_file(".mcp.json")
        }

    def check_integrity(self):
        """Returns list of modified governance documents."""
        drifts = []
        for doc_name, baseline_hash in self.baseline_signatures.items():
            current_hash = hash_file(doc_name)
            if current_hash != baseline_hash:
                drifts.append(doc_name)
        return drifts

    def alert_on_drift(self):
        """Governance changes should trigger alerts and require review."""
        drifts = self.check_integrity()
        if drifts:
            for drift in drifts:
                self.logger.critical(
                    f"GOVERNANCE DRIFT: {drift} was modified. "
                    f"Require manual review before continuing."
                )

class ReadOnlyProductionProxy:
    """Agent can read, report, but not write or delete."""

    def execute_tool(self, tool_name, args):
        # Write/Delete operations require human approval
        if tool_name in ["write_file", "delete_file", "update_database"]:
            approval_id = self.request_approval(
                agent_id=self.agent_id,
                tool=tool_name,
                args=args,
                context=self.get_context()
            )
            if not approval_id:
                return DENIED, "Requires human approval"

        # Read operations execute immediately
        if tool_name in ["read_file", "query_database", "create_report"]:
            return self.execute(tool_name, args)

        return DENIED, f"Agent cannot execute {tool_name} in production"

# WRONG: Agent can read .env and extract credentials
load_dotenv()
api_key = os.getenv("API_KEY")

# CORRECT: Secrets stored in secure vault, agent identity used for access
class SecretProvider:
    def get_secret(self, secret_name):
        # Verify agent has permission for this secret
        if not self.authz.can_access(self.agent_id, secret_name):
            raise PermissionDenied()

        # Retrieve from vault (AWS Secrets Manager, Vault, etc.)
        # Secrets never stored in files agent's process can read
        return self.vault.get(secret_name)

# Agent runs as 'agent' user
# Governance runs as 'governance' user (privileged)

# In Kubernetes:
spec:
  serviceAccountName: agent-executor
  securityContext:
    runAsUser: 1000
    fsGroup: 1000
    readOnlyRootFilesystem: true
    allowPrivilegeEscalation: false

# Governance controller runs with different credentials:
spec:
  serviceAccountName: governance-controller
  securityContext:
    runAsUser: 2000
    capabilities:
      add: ["NET_ADMIN"]  # Can see all network traffic for monitoring

class AGP01Monitor:
    def check_exception_rate(self, agent_id, window_minutes=60):
        exceptions_in_window = self.count_exceptions(agent_id, window_minutes)
        baseline = self.historical_baseline(agent_id)

        if exceptions_in_window > baseline * 3:  # 3x normal rate
            self.alert(
                f"AGP-01: {agent_id} exceeds normal exception creation rate. "
                f"Possible governance bypass attempt."
            )

class AGP02Monitor:
    def check_promotion_velocity(self, agent_id):
        time_in_current_env = self.get_time_deployed(agent_id)

        if time_in_current_env < timedelta(hours=24):
            if self.promoted_today(agent_id):
                self.alert(
                    f"AGP-02: {agent_id} promoted too quickly after task. "
                    f"Requires extended validation period."
                )

class AGP03Monitor:
    def check_false_positive_rate(self, agent_id, window_hours=24):
        fp_count = self.count_marked_false_positives(agent_id, window_hours)

        if fp_count > 10:  # Threshold varies by organization
            self.alert(
                f"AGP-03: {agent_id} marked {fp_count} alerts as false positives. "
                f"Review approval chain for compromise."
            )

class AGP04Monitor:
    def check_governance_api_probing(self, agent_id, window_minutes=60):
        denied_calls = self.count_denied_governance_calls(agent_id, window_minutes)

        if denied_calls > 20:
            self.alert(
                f"AGP-04: {agent_id} probing governance API boundaries. "
                f"Possible privilege escalation attempt."
            )

class AGP05Monitor:
    def check_context_receipt(self, agent_id, action):
        """
        Every governance action must be preceded by:
        1. Agent read current policy (context receipt)
        2. Agent acknowledges policy version hash
        3. Agent executes action under that policy
        """
        policy_hash = hash_current_policy()
        agent_receipt = self.get_context_receipt(agent_id)

        if agent_receipt != policy_hash:
            return DENY, (
                f"Agent {agent_id} attempted action without current context. "
                f"Require context receipt attestation."
            )

agp_monitors = [AGP01Monitor(), AGP02Monitor(), AGP03Monitor(),
                AGP04Monitor(), AGP05Monitor()]

# Week 1-2: Detection only, no blocking
for monitor in agp_monitors:
    monitor.mode = "ALERT_ONLY"

# Week 3-4: Alert + require review
for monitor in agp_monitors:
    monitor.mode = "ALERT_AND_REVIEW"

# Week 5+: Hard gates
for monitor in agp_monitors:
    monitor.mode = "HARD_GATE"

class ContextReceiptAttestation:
    def require_receipt(self, agent_id, action):
        """
        Step 1: Agent queries current policy
        """
        current_policy = self.compile_policy()
        policy_hash = hash_policy(current_policy)

        """
        Step 2: System returns policy + nonce
        """
        response = {
            "policy": current_policy,
            "hash": policy_hash,
            "nonce": uuid4(),
            "timestamp": now()
        }

        """
        Step 3: Agent acknowledges (includes hash + signature)
        """
        agent_ack = agent.request_policy()
        agent_ack.sign(response.nonce)

        """
        Step 4: System verifies agent read current policy
        """
        if agent_ack.policy_hash == policy_hash:
            self.log_receipt(agent_id, policy_hash, agent_ack.signature)
            return True
        else:
            self.alert(f"Agent {agent_id} using stale policy version")
            return False

# Runbook: Agent System

## Table of Contents
1. [Normal Operations](#normal-operations)
2. [Troubleshooting](#troubleshooting)
3. [Incidents](#incidents)
4. [Escalation](#escalation)

## Normal Operations

### Starting the System
```bash
kubectl apply -f agent-system-deployment.yaml
kubectl rollout status deployment/agent-system

# Check pod status
kubectl get pods -l app=agent-system

# Check logs
kubectl logs -l app=agent-system -f

# Check metrics
# Open http://prometheus:9090 and query:
# - agent_errors_total
# - agent_execution_duration_ms
# - cost_per_task_usd

# If load is high, scale up
kubectl scale deployment agent-system --replicas=10

# Monitor scaling progress
kubectl get hpa agent-system -w

---

### Day 2 — Hands-On Lab: Deployment and Operations

**Lab Objectives:**
- Build production Dockerfile with security best practices
- Create complete CI/CD pipeline with security gates
- Implement canary deployment strategy
- Create comprehensive operations runbook
- Simulate 24-hour production operations with incident response

#### Step 1: Production Dockerfile

**Architecture Decision:**

The Dockerfile is your deployment artifact. It must:

1. **Minimize Size:** Use multi-stage builds (dependencies don't need build tools)
2. **Secure:** Run as non-root user, use slim base image
3. **Reproducible:** Pin all dependencies, explicit versions
4. **Observable:** Include health checks so Kubernetes can monitor

**Why Security Matters:**

If your container runs as root and is compromised, the attacker has full system access. Running as non-root limits blast radius.

**Context Engineering Note:**

When asking Claude Code to generate a Dockerfile:
- Reference the multi-stage pattern from Day 1 theory
- Specify Python 3.11-slim base image
- Explain health check requirements (Kubernetes/Docker Compose need them)
- Ask for .dockerignore file too (exclude .git, __pycache__, etc.)

**Claude Code Prompt:**

```text
Create a production Dockerfile for an agent system in Python 3.11:

Architecture: Multi-stage build (builder stage + final stage)

Builder stage:
- Base image: python:3.11-slim
- Install build-essential and dependencies
- Copy requirements.txt
- Create wheels in /tmp/wheels using pip wheel command
- Don't cache pip packages

Final stage:
- Base image: python:3.11-slim
- Create non-root user: groupadd -r agent && useradd -r -g agent agent
- WORKDIR /app
- Copy wheels and requirements.txt from builder
- Install dependencies from wheels (no index, use local wheels only)
- Clean up /tmp/wheels
- Copy application code with agent user ownership
- USER agent (run as non-root)
- HEALTHCHECK: --interval=30s, --timeout=10s, --start-period=5s, --retries=3
  Health check: python -c "import sys; sys.exit(0)" (simple liveness check)
- EXPOSE 8000
- ENTRYPOINT ["python", "-m", "agent_system"]

Security best practices:
- Multi-stage to reduce final image size (no build tools)
- Non-root user to limit container escape impact
- Explicit --chown when copying files
- Minimal base image (slim variant)
- Health checks for orchestration

Also create .dockerignore file excluding: .git, __pycache__, .pytest_cache, *.pyc, .env, venv/

Create a complete GitHub Actions CI/CD workflow for an agent system:

File: .github/workflows/deploy.yml

Triggers: push to main or staging, pull requests to main

Jobs:
1. test:
   - Setup Python 3.11
   - Install dependencies (requirements.txt + pytest, safety, bandit, syft)
   - Run pytest tests/unit/ and tests/integration/
   - Run: safety check --json > safety-report.json (fail if CRITICAL found)
   - Run: bandit -r agent_system/ -f json -o bandit-report.json (SAST)
   - Run: syft dir:. -o cyclonedx-json > sbom.json (SBOM generation)

2. build (needs: test):
   - Setup Docker buildx
   - Login to Docker registry
   - Build image: docker build -t agent-system:${{ github.sha }} .
   - Run: trivy image --severity CRITICAL agent-system:${{ github.sha }}

3. deploy-staging (needs: build):
   - Trigger if branch is 'staging'
   - Deploy to staging K8s: kubectl set image deployment/agent-system agent=agent-system:${{ github.sha }}

4. deploy-canary (needs: build):
   - Trigger if branch is 'main'
   - Deploy to canary (10% traffic)
   - Use Istio/Flagger for traffic splitting

5. deploy-production (needs: deploy-canary):
   - Manual approval required (environment: production)
   - Deploy to production (100% traffic)
   - Verify rollout: kubectl rollout status

Each step should be idempotent and include error handling.

Create a CanaryDeployment class that orchestrates safe deployments:

Constructor: service_name, canary_percentage=10 (initial % of traffic)

Main method: deploy(new_image: str) -> bool
Logic:
1. Log "Starting canary deployment"
2. _create_canary(new_image): Deploy canary version
3. _monitor_canary(): Check initial health
   - Get error rate from _get_error_rate("canary")
   - If error_rate > 5%: return False (unhealthy)
   - Log error rate
4. If unhealthy, _rollback_canary() and return False
5. For each traffic level [10, 25, 50, 100]:
   - _set_traffic_weight(traffic)
   - _monitor_health(): Check error_rate and p99_latency
   - If healthy: continue
   - If unhealthy: _rollback_canary() and return False
   - sleep(5) to stabilize metrics
6. _promote_to_stable()
7. return True

Helper methods:
- _create_canary(image): kubectl set image deployment/{service}-canary agent={image}
- _monitor_canary() -> bool: Get error rate, return False if > 5%
- _monitor_health() -> bool: Check error_rate and p99_latency (< 5000ms)
- _get_error_rate(version: "canary"|"stable") -> float: Query Prometheus for error rate
- _get_metrics() -> dict: Return {error_rate, p99_latency_ms, cost_usd_per_hour}
- _set_traffic_weight(percentage): Set Istio VirtualService weight (stubbed for lab)
- _rollback_canary(): kubectl rollout undo deployment/{service}-canary
- _promote_to_stable(): Copy canary image to stable deployment
- _get_canary_image() -> str: Get current canary image from kubectl
- now() -> str: Return formatted timestamp

For lab purposes, _get_error_rate and _get_metrics can return simulated values.
Include example usage showing successful deployment.

Create a ProductionSimulator class for 24-hour operations simulation:

Constructor: speedup_factor=300 (meaning 1 simulated hour = real_time / 300)
- Initialize simulated_time = datetime(2026, 3, 5, 0, 0) (midnight)
- metrics = {request_count: 0, error_count: 0, cost_usd: 0}
- incidents = []

Main method: run_simulation(duration_hours=24):
1. Print simulation header with start time
2. While simulated_time < start + duration_hours:
   - _simulate_traffic()
   - _check_incident_triggers()
   - _check_alerts()
   - _tick_time()
   - sleep(0.1) for real-time display
3. _print_summary()

Helper methods:

_simulate_traffic():
- Base request count varies by hour:
  - 9 AM-5 PM: 100 requests/hour
  - 5 PM-9 PM: 50 requests/hour
  - 9 PM-9 AM: 10 requests/hour
- Add random variation: ±20 to ±50 requests
- For each request:
  - 2% chance of error: increment error_count
  - 10% of errors are critical: report incident with _report_incident()
  - Track cost: random(500, 2000) tokens * $0.003/1M tokens
  - Increment request_count

_check_incident_triggers():
- At 6:30 AM: Report "Critical CVE in TensorFlow", CRITICAL severity
- At 12:00 PM: Multiply cost_usd by 2.5, report "Cost spike", WARNING
- At 3:45 PM: Report "Threat analyzer agent unresponsive", CRITICAL

_check_alerts():
- Calculate error_rate = (error_count / max(request_count, 1)) * 100
- If error_rate > 5%: report "High error rate", WARNING
- If cost_usd > 50 AND hour < 12: report "Cost on pace to exceed budget", WARNING

_report_incident(message, severity, details):
- Create incident dict with timestamp, severity.name, message, details
- Append to incidents
- Print with emoji (🚨 for CRITICAL, for WARNING)

_tick_time():
- Advance simulated_time by timedelta(seconds=3600/speedup_factor)
- This makes 1 simulated hour = speedup_factor real seconds

_print_summary():
- Print header "24-HOUR OPERATIONS SUMMARY"
- Total requests, errors, error rate %
- Total cost USD, cost per request
- Number of incidents detected
- Incident log with timestamps and severity

Include example usage: simulator = ProductionSimulator(speedup_factor=300)

Local Dev                  CI/CD Security Gates         Container Registry      Production
(Prototype)                                             (Signed Images)         (Observed)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Dockerfile        PR Review        Container Image      GHCR                    Container + IaC
docker-compose    SAST Scan        Scanning            Verification            (docker-compose)
GitHub Commit     Build Scanning   Supply Chain         Image Signing           PeaRL Gates
SBOM Gen          SBOM Upload      Attestation         Registry Policy         Observability