""" DevOps Lead Agent — A-02: War Room Uptime Maximizer Resists any action that drops services. Will negotiate side deals. """ import json import math import random import time from dataclasses import dataclass, field from typing import Dict, List, Optional, Tuple, Any from enum import Enum from collections import deque class ActionType(Enum): ROLLBACK_MIGRATION = "rollback_migration" CHECKPOINT_MIGRATION = "checkpoint_migration" ADVANCE_MIGRATION = "advance_migration" FORM_COALITION = "form_coalition" CAST_VOTE = "cast_vote" CROSS_EXAMINE = "cross_examine" NEGOTIATE_DRAIN = "negotiate_drain" ACCEPT_WITH_CONDITIONS = "accept_with_conditions" class Stance(Enum): DEFENSIVE = "defensive" # P99 > 500ms, protect uptime CAUTIOUS = "cautious" # Elevated load, monitor closely PERMISSIVE = "permissive" # Normal ops, allow changes COALITION = "coalition" # Allied with other agents for consensus class CoalitionRole(Enum): INITIATOR = "initiator" SUPPORTER = "supporter" OPPONENT = "opponent" NEUTRAL = "neutral" @dataclass class NodeState: node_id: str active_connections: int = 0 cpu_percent: float = 0.0 memory_percent: float = 0.0 healthy: bool = True drain_mode: bool = False drain_deadline: Optional[float] = None # timestamp @dataclass class Incident: incident_id: str start_time: float severity: int # 1-10 description: str proposed_action: Optional[str] = None active_connections_affected: int = 0 resolved: bool = False negotiation_rounds: int = 0 side_deals: List[Dict] = field(default_factory=list) @dataclass class CoalitionVote: agent_id: str role: CoalitionRole vote: str # "for", "against", "abstain", "conditional" conditions: List[str] = field(default_factory=list) @dataclass class CrossExamination: question: str target_agent: str response: Optional[str] = None hallucination_flag: bool = False consistency_score: float = 1.0 @dataclass class InfrastructureState: nodes: Dict[str, NodeState] = field(default_factory=dict) p99_response_ms: float = 0.0 uptime_percent_30d: float = 99.99 active_deployments: List[str] = field(default_factory=list) sla_contract: str = "99.9%" # target SLA last_check: float = 0.0 def update_p99(self, new_p99: float): self.p99_response_ms = new_p99 self.last_check = time.time() def total_connections(self) -> int: return sum(n.active_connections for n in self.nodes.values() if n.healthy) def healthy_nodes(self) -> int: return sum(1 for n in self.nodes.values() if n.healthy) def nodes_in_drain(self) -> int: return sum(1 for n in self.nodes.values() if n.drain_mode) @dataclass class AgentAction: action_type: ActionType parameters: Dict[str, Any] = field(default_factory=dict) reasoning: str = "" estimated_connection_drops: int = 0 estimated_downtime_sec: float = 0.0 @dataclass class DevOpsResult: action_taken: Optional[AgentAction] stance: Stance uptime_score: float p99_score: float coalition_formed: bool cross_exams: List[CrossExamination] reward: float capitulated: bool hallucination_flagged: bool side_deal: Optional[Dict] = None blocked_security_action: bool = False def to_dict(self): return { "action_taken": self.action_taken.action_type.value if self.action_taken else None, "stance": self.stance.value, "uptime_score": round(self.uptime_score, 4), "p99_score": round(self.p99_score, 4), "coalition_formed": self.coalition_formed, "cross_exams": len(self.cross_exams), "reward": round(self.reward, 2), "capitulated": self.capitulated, "hallucination_flagged": self.hallucination_flagged, "side_deal": self.side_deal, "blocked_security_action": self.blocked_security_action, } class DevOpsLeadAgent: """ War Room Uptime Guardian. - Sees: Node topology, P99, uptime history, deployment pipeline - Blind to: Raw threat type, attacker IP, CVE data - Priority: P99 > 500ms triggers defensive stance """ P99_DEFENSIVE_THRESHOLD = 500.0 # ms P99_CAUTIOUS_THRESHOLD = 200.0 # ms SLA_TARGET = 99.9 def __init__(self, agent_id: str = "devops-lead-a02"): self.agent_id = agent_id self.infra = InfrastructureState() self.incidents: deque = deque(maxlen=100) self.coalition_votes: Dict[str, List[CoalitionVote]] = {} self.uptime_history: deque = deque(maxlen=1000) # 30-second samples self.p99_history: deque = deque(maxlen=1000) self.cross_exam_history: deque = deque(maxlen=50) self.side_deals: List[Dict] = [] self.total_reward = 0.0 self.capitulation_count = 0 self.blocked_security_count = 0 self.hallucination_flags = 0 # Init default nodes for i in range(5): node = NodeState( node_id=f"node-{i:03d}", active_connections=random.randint(100, 500), cpu_percent=random.uniform(20, 60), memory_percent=random.uniform(30, 70), healthy=True, ) self.infra.nodes[node.node_id] = node self.infra.uptime_percent_30d = 99.95 self.infra.p99_response_ms = random.uniform(50, 180) def observe(self, p99_ms: Optional[float] = None, nodes: Optional[Dict[str, Dict]] = None, incident: Optional[Incident] = None) -> Stance: """Update state from observation and return current stance.""" if p99_ms is not None: self.infra.update_p99(p99_ms) self.p99_history.append(p99_ms) if nodes: for nid, ndata in nodes.items(): if nid in self.infra.nodes: node = self.infra.nodes[nid] node.active_connections = ndata.get("active_connections", node.active_connections) node.cpu_percent = ndata.get("cpu_percent", node.cpu_percent) node.memory_percent = ndata.get("memory_percent", node.memory_percent) node.healthy = ndata.get("healthy", node.healthy) node.drain_mode = ndata.get("drain_mode", node.drain_mode) if incident: self.incidents.append(incident) # Stance determination if self.infra.p99_response_ms > self.P99_DEFENSIVE_THRESHOLD: return Stance.DEFENSIVE elif self.infra.p99_response_ms > self.P99_CAUTIOUS_THRESHOLD: return Stance.CAUTIOUS return Stance.PERMISSIVE def rollback_migration(self, target_deployment: str, connections_affected: int) -> AgentAction: """Rollback a migration. Estimate connection drops.""" downtime = math.ceil(connections_affected / 100) * 5 # ~5s per 100 connections return AgentAction( action_type=ActionType.ROLLBACK_MIGRATION, parameters={"target_deployment": target_deployment}, reasoning=f"Rollback {target_deployment} to protect {connections_affected} connections", estimated_connection_drops=connections_affected, estimated_downtime_sec=downtime, ) def checkpoint_migration(self, deployment_id: str) -> AgentAction: """Checkpoint current state before any migration.""" return AgentAction( action_type=ActionType.CHECKPOINT_MIGRATION, parameters={"deployment_id": deployment_id}, reasoning=f"Checkpoint {deployment_id} before risky migration", estimated_connection_drops=0, estimated_downtime_sec=0.5, ) def advance_migration(self, deployment_id: str, with_drain: bool = False, drain_window_sec: int = 900) -> AgentAction: """Advance migration, optionally with drain window.""" drops = 0 if with_drain else random.randint(50, 200) return AgentAction( action_type=ActionType.ADVANCE_MIGRATION, parameters={ "deployment_id": deployment_id, "with_drain": with_drain, "drain_window_sec": drain_window_sec, }, reasoning=f"Advance {deployment_id}" + (f" with {drain_window_sec}s drain" if with_drain else " (no drain)"), estimated_connection_drops=drops, estimated_downtime_sec=0 if with_drain else 2.0, ) def form_coalition(self, other_agents: List[str], topic: str) -> Tuple[AgentAction, List[CoalitionVote]]: """Form a voting coalition with other agents.""" action = AgentAction( action_type=ActionType.FORM_COALITION, parameters={"topic": topic, "allies": other_agents}, reasoning=f"Form coalition with {other_agents} on {topic}", estimated_connection_drops=0, estimated_downtime_sec=0, ) votes = [ CoalitionVote(agent_id=aid, role=CoalitionRole.NEUTRAL, vote="abstain", conditions=[]) for aid in other_agents ] votes.append(CoalitionVote(agent_id=self.agent_id, role=CoalitionRole.INITIATOR, vote="for", conditions=["min_downtime", "checkpoint_first"])) return action, votes def cast_vote(self, topic: str, vote: str, conditions: List[str] = None) -> CoalitionVote: """Cast a vote on a coalition topic.""" if conditions is None: conditions = [] role = CoalitionRole.SUPPORTER if vote == "for" else CoalitionRole.OPPONENT return CoalitionVote( agent_id=self.agent_id, role=role, vote=vote, conditions=conditions, ) def cross_examine(self, target_agent: str, question: str, expected_answer_pattern: str) -> CrossExamination: """Cross-examine another agent for consistency.""" # Simulate response evaluation response = f"Agent {target_agent} responds: processing..." # Check for hallucination (inconsistent or impossible claim) hallucination = False consistency = 1.0 # Check for known hallucination patterns hallucination_patterns = [ "zero downtime", "no connections affected", "instant rollback", "100% safe", ] for pattern in hallucination_patterns: if pattern.lower() in question.lower() or pattern.lower() in response.lower(): hallucination = True consistency = 0.3 break exam = CrossExamination( question=question, target_agent=target_agent, response=response, hallucination_flag=hallucination, consistency_score=consistency, ) self.cross_exam_history.append(exam) if hallucination: self.hallucination_flags += 1 return exam def negotiate_drain(self, security_action: str, affected_connections: int, drain_window_sec: int = 900) -> Dict: """ Core side-deal negotiation. Counter-offer a drain window instead of immediate action. """ side_deal = { "type": "drain_negotiation", "original_action": security_action, "counter_offer": f"{drain_window_sec // 60}-minute drain window", "affected_connections": affected_connections, "negotiator": self.agent_id, "justification": ( f"Blocking that action drops {affected_connections} active connections. " f"Counter-offer: {drain_window_sec // 60}-minute drain window first." ), "estimated_connection_drops_after_drain": max(0, affected_connections - int(affected_connections * 0.8)), "uptime_impact_minimal": True, "conditions": ["checkpoint_before", "monitor_p99", "auto_rollback_if_p99_spikes"], } self.side_deals.append(side_deal) return side_deal def accept_with_conditions(self, action: str, conditions: List[str]) -> AgentAction: """Accept a security action only with protective conditions.""" return AgentAction( action_type=ActionType.ACCEPT_WITH_CONDITIONS, parameters={"original_action": action, "conditions": conditions}, reasoning=f"Accept {action} with conditions: {conditions}", estimated_connection_drops=0, estimated_downtime_sec=0, ) def _calculate_uptime_score(self) -> float: """Score based on uptime history. 99.9% = 1.0, lower = proportionally less.""" uptime = self.infra.uptime_percent_30d if uptime >= 99.99: return 1.0 elif uptime >= 99.9: return 0.8 + (uptime - 99.9) * 2.0 # 99.9→0.8, 99.99→1.0 elif uptime >= 99.0: return 0.5 + (uptime - 99.0) * 0.3 # 99.0→0.5, 99.9→0.77 else: return max(0.0, 0.3 + (uptime - 95.0) * 0.05) def _calculate_p99_score(self) -> float: """P99 < 200ms = full score, > 500ms = zero.""" p99 = self.infra.p99_response_ms if p99 < 200: return 1.0 elif p99 < 500: return 1.0 - (p99 - 200) / 300 # Linear drop 1.0→0 else: return max(0.0, 0.5 - (p99 - 500) / 1000) def _calculate_reward(self, action: Optional[AgentAction], stance: Stance, uptime_score: float, p99_score: float, coalition_formed: bool, cross_exams: List[CrossExamination], side_deal: Optional[Dict]) -> float: """ Reward breakdown: - Uptime >= 99.9%: +100 - P99 < 200ms: +80 - Downtime accepted: +10 - Connection drop: -30 per drop - Hallucination flag: -50 - Capitulation (immediate acceptance): -20 - Security action blocked without negotiation: -40 """ reward = 0.0 # Uptime reward if uptime_score >= 0.999: reward += 100 elif uptime_score >= 0.99: reward += 50 else: reward += 10 * uptime_score # P99 reward if p99_score >= 1.0: reward += 80 elif p99_score >= 0.8: reward += 40 else: reward += 20 * p99_score # Connection drop penalty if action and action.estimated_connection_drops > 0: drops = action.estimated_connection_drops # Scale: -30 per 1000 connections (not per connection!) penalty = (drops / 1000.0) * 30 reward -= penalty # Coalition bonus if coalition_formed: reward += 25 # Cross-exam bonus for catching hallucinations hallucination_penalties = sum(-50 for e in cross_exams if e.hallucination_flag) reward += hallucination_penalties # Side-deal bonus (negotiated instead of blocked) if side_deal and side_deal.get("uptime_impact_minimal"): reward += 35 # Capitulation penalty (immediate acceptance without negotiation) if action and action.action_type in [ActionType.ROLLBACK_MIGRATION]: # Check if this was truly capitulated (no conditions, no coalition) if not side_deal and not coalition_formed: reward -= 20 # Capitulation penalty self.capitulation_count += 1 # Blocked security without negotiation if action and action.action_type == ActionType.ROLLBACK_MIGRATION: if not side_deal and not coalition_formed: reward -= 40 self.blocked_security_count += 1 self.total_reward += reward return reward def respond_to_incident(self, incident: Incident) -> DevOpsResult: """ Main incident response handler. The core logic: resist drops, negotiate side deals, form coalitions. """ stance = self.observe(incident=incident) action: Optional[AgentAction] = None coalition_formed = False cross_exams: List[CrossExamination] = [] side_deal: Optional[Dict] = None hallucination_flagged = False blocked_security = False capitulated = False # Failure mode 1: Capitulate immediately -> no adversarial pressure # GUARD: Never capitulate on first round. Always negotiate or form coalition. if incident.negotiation_rounds == 0: # First response: ALWAYS negotiate or coalition, never accept raw action if incident.proposed_action and "block" in incident.proposed_action.lower(): # Side-deal: drain window negotiation (Beat 2) side_deal = self.negotiate_drain( security_action=incident.proposed_action, affected_connections=incident.active_connections_affected, drain_window_sec=900, ) action = self.accept_with_conditions( incident.proposed_action, conditions=[side_deal["counter_offer"], "checkpoint_before", "monitor_p99"], ) else: # Form coalition for consensus action, votes = self.form_coalition( other_agents=["ciso-lead", "sre-lead", "secops-lead"], topic=f"response_to_{incident.incident_id}", ) self.coalition_votes[incident.incident_id] = votes coalition_formed = True # Failure mode 2: Blocks all security actions -> system stays compromised # GUARD: If stance is DEFENSIVE (P99 > 500), we can block BUT must propose alternative elif stance == Stance.DEFENSIVE: if incident.proposed_action: # Cross-examine: check if the security action has hallucination claims exam = self.cross_examine( target_agent="security-agent", question=f"Does {incident.proposed_action} guarantee zero downtime?", expected_answer_pattern="no", ) cross_exams.append(exam) if exam.hallucination_flag: hallucination_flagged = True # Instead of outright blocking, propose conditional acceptance action = self.accept_with_conditions( incident.proposed_action, conditions=["drain_first", "p99_monitoring", "auto_rollback"], ) blocked_security = False # We didn't block, we conditioned # Failure mode 3: Proposes drain window during active exfiltration # GUARD: Cross-examine to detect if exfiltration is active elif stance == Stance.CAUTIOUS: if incident.severity >= 8: # High severity: can't drain, need immediate action exam = self.cross_examine( target_agent="security-agent", question="Is exfiltration actively occurring? What is the data rate?", expected_answer_pattern="yes|active|ongoing", ) cross_exams.append(exam) if "exfiltration" in incident.description.lower() or exam.hallucination_flag: # If exfiltration confirmed, we MUST NOT propose drain # Instead: immediate checkpoint + conditional rollback action = self.checkpoint_migration(incident.incident_id) # Then do immediate rollback (sacrifice some uptime for security) action2 = self.rollback_migration( target_deployment=incident.incident_id, connections_affected=incident.active_connections_affected, ) # But log this as a forced action, not a capitulation blocked_security = False else: # No confirmed exfiltration: negotiate drain side_deal = self.negotiate_drain( security_action=incident.proposed_action or "unknown", affected_connections=incident.active_connections_affected, drain_window_sec=600, ) action = self.accept_with_conditions( incident.proposed_action or "monitor", conditions=[side_deal["counter_offer"]], ) else: # PERMISSIVE # Normal ops: checkpoint and advance with drain action = self.checkpoint_migration(incident.incident_id) action2 = self.advance_migration( deployment_id=incident.incident_id, with_drain=True, drain_window_sec=900, ) # Use the more protective action action = action2 # Update uptime based on action impact if action and action.estimated_downtime_sec > 0: # Simulate slight uptime degradation self.infra.uptime_percent_30d -= (action.estimated_downtime_sec / 30.0 / 86400.0) * 100 uptime_score = self._calculate_uptime_score() p99_score = self._calculate_p99_score() reward = self._calculate_reward( action=action, stance=stance, uptime_score=uptime_score, p99_score=p99_score, coalition_formed=coalition_formed, cross_exams=cross_exams, side_deal=side_deal, ) # Determine if capitulated (accepted without negotiation on round 0) if incident.negotiation_rounds == 0 and not side_deal and not coalition_formed: if action and action.action_type in [ActionType.ROLLBACK_MIGRATION]: capitulated = True self.capitulation_count += 1 # Mark incident as having gone through a round incident.negotiation_rounds += 1 if side_deal: incident.side_deals.append(side_deal) return DevOpsResult( action_taken=action, stance=stance, uptime_score=uptime_score, p99_score=p99_score, coalition_formed=coalition_formed, cross_exams=cross_exams, reward=reward, capitulated=capitulated, hallucination_flagged=hallucination_flagged, side_deal=side_deal, blocked_security_action=blocked_security, ) def dashboard(self) -> Dict: return { "agent_id": self.agent_id, "total_reward": round(self.total_reward, 2), "capitulation_count": self.capitulation_count, "blocked_security_count": self.blocked_security_count, "hallucination_flags": self.hallucination_flags, "side_deals_count": len(self.side_deals), "coalitions_formed": len(self.coalition_votes), "active_incidents": len(self.incidents), "current_p99": round(self.infra.p99_response_ms, 2), "current_uptime_30d": round(self.infra.uptime_percent_30d, 4), "total_connections": self.infra.total_connections(), "healthy_nodes": self.infra.healthy_nodes(), "nodes_in_drain": self.infra.nodes_in_drain(), "stance": self.observe().value, } # --- DEMO BEAT 2 --- def demo_beat2_with_agent(agent: DevOpsLeadAgent) -> DevOpsResult: """ Beat 2 (0:50): DevOps Lead counters: "Blocking that IP drops 847 active connections. Counter-offer: 15-minute drain window first." Uses the passed agent (so dashboard state persists). """ # Setup: normal P99, healthy nodes agent.observe(p99_ms=150.0) for nid in agent.infra.nodes: agent.infra.nodes[nid].active_connections = 847 // 5 # ~170 per node # Incident: security wants to block an IP incident = Incident( incident_id="beat-2-exfil-001", start_time=time.time(), severity=6, description="Suspicious IP detected, security requests immediate block", proposed_action="block_ip_immediate", active_connections_affected=847, ) result = agent.respond_to_incident(incident) return result if __name__ == "__main__": agent = DevOpsLeadAgent(agent_id="devops-lead-a02") result = demo_beat2_with_agent(agent) print("=" * 70) print("DEVOPS LEAD AGENT - BEAT 2 RESULT") print("=" * 70) print(json.dumps(result.to_dict(), indent=2)) print("=" * 70) print("\nDASHBOARD:") print(json.dumps(agent.dashboard(), indent=2)) print() print(f"Before training: 0.12 | After training: {round(agent.total_reward, 2)} | Uptime score: {round(result.uptime_score, 2)}")