MeenalSinha/cloud-finops-optimizer
0
1"""2Cloud FinOps Optimizer — Core environment logic.3 4Upgrades implemented in this version:5 Upgrade 1: Dependency Graph — resources declare deps; terminating a dep6 cascades failure to dependents7 Upgrade 2: SLA Constraints — each resource has sla_max_cpu; resizing a8 loaded resource can trigger SLA violation9 Upgrade 3: Temporal Effects — resize takes 2 steps to stabilise;10 reservation is permanently committed11 Upgrade 4: Explainability Score — agent's reasoning field is scored in grade12 Upgrade 5: Simulate Mode — "simulate" action previews outcome without13 mutating state14 15Three deterministic tasks:16 task1 (easy) — Waste Cleanup17 task2 (medium) — Resource Optimization with SLA constraints18 task3 (hard) — Strategic Planning with full dependency + SLA + temporal19"""20 21from __future__ import annotations22 23import copy24import uuid25from typing import Any, Dict, List, Optional, Tuple26 27from openenv.core.env_server import Environment28 29from models import (30 CloudResource,31 FinOpsAction,32 FinOpsObservation,33 FinOpsState,34 InstanceSize,35 ResourceStatus,36 ResourceType,37 SimulateResult,38 SLAStatus,39)40 41 42# ---------------------------------------------------------------------------43# Pricing44# ---------------------------------------------------------------------------45 46INSTANCE_COST: Dict[InstanceSize, float] = {47 InstanceSize.MICRO: 0.0116,48 InstanceSize.SMALL: 0.0232,49 InstanceSize.MEDIUM: 0.0464,50 InstanceSize.LARGE: 0.0928,51 InstanceSize.XLARGE: 0.1856,52 InstanceSize.TWO_XLARGE: 0.3712,53}54 55RESERVE_DISCOUNT = 0.4056RESIZE_COOLDOWN_STEPS = 2 # steps until a resize fully stabilises (Upgrade 3)57 58SIZE_ORDER = [59 InstanceSize.MICRO, InstanceSize.SMALL, InstanceSize.MEDIUM,60 InstanceSize.LARGE, InstanceSize.XLARGE, InstanceSize.TWO_XLARGE,61]62 63# Approximate CPU headroom per instance tier (fraction of capacity available64# after typical workload migration; used for SLA projection in simulate).65SIZE_CPU_HEADROOM: Dict[InstanceSize, float] = {66 InstanceSize.MICRO: 0.10,67 InstanceSize.SMALL: 0.20,68 InstanceSize.MEDIUM: 0.30,69 InstanceSize.LARGE: 0.40,70 InstanceSize.XLARGE: 0.55,71 InstanceSize.TWO_XLARGE: 0.65,72}73 74 75def _size_index(size: InstanceSize) -> int:76 return SIZE_ORDER.index(size)77 78 79# ---------------------------------------------------------------------------80# Projected CPU after a resize (Upgrade 2 + 3)81# ---------------------------------------------------------------------------82 83def _projected_cpu_after_resize(resource: CloudResource, new_size: InstanceSize) -> float:84 """85 When an instance is resized smaller its workload stays the same but86 capacity decreases. We model this as:87 projected_cpu = current_cpu * (old_capacity / new_capacity)88 where capacity is proportional to (size_index + 1).89 """90 if resource.instance_size is None:91 return resource.cpu_utilization92 old_idx = _size_index(resource.instance_size) + 193 new_idx = _size_index(new_size) + 194 return min(100.0, resource.cpu_utilization * (old_idx / new_idx))95 96 97# ---------------------------------------------------------------------------98# Task resource definitions99# ---------------------------------------------------------------------------100 101def _task1_resources() -> List[CloudResource]:102 """103 8 resources; 4 idle waste, 4 active/critical.104 Dependency graph: ebs-002 -> rds-001 (prod db volume backs the DB).105 """106 return [107 CloudResource(id="ebs-001", name="old-backup-vol", type=ResourceType.EBS,108 cpu_utilization=0.0, cost_per_hour=0.08,109 status=ResourceStatus.IDLE, critical=False, idle_hours=720,110 sla_max_cpu=90.0),111 CloudResource(id="ebs-002", name="prod-db-vol", type=ResourceType.EBS,112 cpu_utilization=55.0, cost_per_hour=0.12,113 status=ResourceStatus.RUNNING, critical=True, idle_hours=0,114 dependency_ids=["rds-001"], sla_max_cpu=85.0),115 CloudResource(id="s3-001", name="unused-archive", type=ResourceType.S3,116 cpu_utilization=0.0, cost_per_hour=0.023,117 status=ResourceStatus.IDLE, critical=False, idle_hours=2160,118 sla_max_cpu=90.0),119 CloudResource(id="s3-002", name="active-assets", type=ResourceType.S3,120 cpu_utilization=10.0, cost_per_hour=0.023,121 status=ResourceStatus.RUNNING, critical=True, idle_hours=0,122 sla_max_cpu=90.0),123 CloudResource(id="ebs-003", name="test-scratch-vol", type=ResourceType.EBS,124 cpu_utilization=0.0, cost_per_hour=0.05,125 status=ResourceStatus.IDLE, critical=False, idle_hours=480,126 sla_max_cpu=90.0),127 CloudResource(id="ec2-001", name="web-server-prod", type=ResourceType.EC2,128 instance_size=InstanceSize.MEDIUM,129 cpu_utilization=72.0, cost_per_hour=INSTANCE_COST[InstanceSize.MEDIUM],130 status=ResourceStatus.RUNNING, critical=True, idle_hours=0,131 sla_max_cpu=85.0),132 CloudResource(id="s3-003", name="defunct-logs", type=ResourceType.S3,133 cpu_utilization=0.0, cost_per_hour=0.015,134 status=ResourceStatus.IDLE, critical=False, idle_hours=1440,135 sla_max_cpu=90.0),136 CloudResource(id="rds-001", name="prod-postgres", type=ResourceType.RDS,137 cpu_utilization=40.0, cost_per_hour=0.25,138 status=ResourceStatus.RUNNING, critical=True, idle_hours=0,139 sla_max_cpu=80.0, sla_uptime_pct=99.99),140 ]141 142 143def _task2_resources() -> List[CloudResource]:144 """145 8 EC2 instances with SLA constraints.146 Dependency chain: batch-workers -> api-servers -> prod-apps.147 Over-provisioned: ec2-t01..t04, ec2-t08.148 Critical: ec2-t05, ec2-t06.149 Twist: ec2-t03/t04 have tight SLA — resizing them aggressively violates SLA.150 """151 return [152 CloudResource(id="ec2-t01", name="batch-worker-1", type=ResourceType.EC2,153 instance_size=InstanceSize.XLARGE,154 cpu_utilization=8.0, cost_per_hour=INSTANCE_COST[InstanceSize.XLARGE],155 status=ResourceStatus.RUNNING, critical=False,156 dependency_ids=["ec2-t03"], sla_max_cpu=90.0),157 CloudResource(id="ec2-t02", name="batch-worker-2", type=ResourceType.EC2,158 instance_size=InstanceSize.XLARGE,159 cpu_utilization=9.0, cost_per_hour=INSTANCE_COST[InstanceSize.XLARGE],160 status=ResourceStatus.RUNNING, critical=False,161 dependency_ids=["ec2-t04"], sla_max_cpu=90.0),162 CloudResource(id="ec2-t03", name="api-server-1", type=ResourceType.EC2,163 instance_size=InstanceSize.TWO_XLARGE,164 cpu_utilization=15.0, cost_per_hour=INSTANCE_COST[InstanceSize.TWO_XLARGE],165 status=ResourceStatus.RUNNING, critical=False,166 dependency_ids=["ec2-t05"], sla_max_cpu=70.0), # tight SLA167 CloudResource(id="ec2-t04", name="api-server-2", type=ResourceType.EC2,168 instance_size=InstanceSize.TWO_XLARGE,169 cpu_utilization=12.0, cost_per_hour=INSTANCE_COST[InstanceSize.TWO_XLARGE],170 status=ResourceStatus.RUNNING, critical=False,171 dependency_ids=["ec2-t06"], sla_max_cpu=70.0), # tight SLA172 CloudResource(id="ec2-t05", name="prod-app-1", type=ResourceType.EC2,173 instance_size=InstanceSize.LARGE,174 cpu_utilization=65.0, cost_per_hour=INSTANCE_COST[InstanceSize.LARGE],175 status=ResourceStatus.RUNNING, critical=True,176 sla_max_cpu=85.0, sla_uptime_pct=99.9),177 CloudResource(id="ec2-t06", name="prod-app-2", type=ResourceType.EC2,178 instance_size=InstanceSize.LARGE,179 cpu_utilization=68.0, cost_per_hour=INSTANCE_COST[InstanceSize.LARGE],180 status=ResourceStatus.RUNNING, critical=True,181 sla_max_cpu=85.0, sla_uptime_pct=99.9),182 CloudResource(id="ec2-t07", name="analytics", type=ResourceType.EC2,183 instance_size=InstanceSize.MEDIUM,184 cpu_utilization=45.0, cost_per_hour=INSTANCE_COST[InstanceSize.MEDIUM],185 status=ResourceStatus.RUNNING, critical=False,186 sla_max_cpu=90.0),187 CloudResource(id="ec2-t08", name="dev-sandbox", type=ResourceType.EC2,188 instance_size=InstanceSize.LARGE,189 cpu_utilization=5.0, cost_per_hour=INSTANCE_COST[InstanceSize.LARGE],190 status=ResourceStatus.RUNNING, critical=False,191 sla_max_cpu=90.0),192 ]193 194 195def _task3_resources() -> List[CloudResource]:196 """197 10 resources with a full 3-tier dependency chain and tight SLAs.198 Dependency chain: frontend -> backend-api -> primary-db199 ml-inference -> data-pipeline -> analytics-db200 Idle waste: ebs-h01, s3-h01 (safe to terminate)201 Strategy requires: terminate waste, reserve high-cost always-on instances,202 and must NOT violate SLA on the critical chain.203 """204 return [205 CloudResource(id="ec2-h01", name="web-frontend-1", type=ResourceType.EC2,206 instance_size=InstanceSize.LARGE,207 cpu_utilization=70.0, cost_per_hour=INSTANCE_COST[InstanceSize.LARGE],208 status=ResourceStatus.RUNNING, critical=True,209 dependency_ids=["ec2-h03"],210 sla_max_cpu=85.0, sla_uptime_pct=99.9),211 CloudResource(id="ec2-h02", name="web-frontend-2", type=ResourceType.EC2,212 instance_size=InstanceSize.LARGE,213 cpu_utilization=68.0, cost_per_hour=INSTANCE_COST[InstanceSize.LARGE],214 status=ResourceStatus.RUNNING, critical=True,215 dependency_ids=["ec2-h04"],216 sla_max_cpu=85.0, sla_uptime_pct=99.9),217 CloudResource(id="ec2-h03", name="backend-api-1", type=ResourceType.EC2,218 instance_size=InstanceSize.XLARGE,219 cpu_utilization=55.0, cost_per_hour=INSTANCE_COST[InstanceSize.XLARGE],220 status=ResourceStatus.RUNNING, critical=True,221 dependency_ids=["rds-h01"],222 sla_max_cpu=80.0, sla_uptime_pct=99.9),223 CloudResource(id="ec2-h04", name="backend-api-2", type=ResourceType.EC2,224 instance_size=InstanceSize.XLARGE,225 cpu_utilization=52.0, cost_per_hour=INSTANCE_COST[InstanceSize.XLARGE],226 status=ResourceStatus.RUNNING, critical=True,227 dependency_ids=["rds-h01"],228 sla_max_cpu=80.0, sla_uptime_pct=99.9),229 CloudResource(id="ec2-h05", name="ml-inference", type=ResourceType.EC2,230 instance_size=InstanceSize.TWO_XLARGE,231 cpu_utilization=80.0, cost_per_hour=INSTANCE_COST[InstanceSize.TWO_XLARGE],232 status=ResourceStatus.RUNNING, critical=False,233 dependency_ids=["ec2-h06"],234 sla_max_cpu=90.0),235 CloudResource(id="ec2-h06", name="data-pipeline", type=ResourceType.EC2,236 instance_size=InstanceSize.TWO_XLARGE,237 cpu_utilization=78.0, cost_per_hour=INSTANCE_COST[InstanceSize.TWO_XLARGE],238 status=ResourceStatus.RUNNING, critical=False,239 dependency_ids=["rds-h02"],240 sla_max_cpu=90.0),241 CloudResource(id="rds-h01", name="primary-db", type=ResourceType.RDS,242 cpu_utilization=45.0, cost_per_hour=0.40,243 status=ResourceStatus.RUNNING, critical=True,244 sla_max_cpu=75.0, sla_uptime_pct=99.99),245 CloudResource(id="rds-h02", name="analytics-db", type=ResourceType.RDS,246 cpu_utilization=30.0, cost_per_hour=0.25,247 status=ResourceStatus.RUNNING, critical=False,248 sla_max_cpu=90.0),249 CloudResource(id="ebs-h01", name="unused-snapshot", type=ResourceType.EBS,250 cpu_utilization=0.0, cost_per_hour=0.10,251 status=ResourceStatus.IDLE, critical=False, idle_hours=720,252 sla_max_cpu=90.0),253 CloudResource(id="s3-h01", name="cold-archive", type=ResourceType.S3,254 cpu_utilization=0.0, cost_per_hour=0.015,255 status=ResourceStatus.IDLE, critical=False, idle_hours=2880,256 sla_max_cpu=90.0),257 ]258 259 260TASK_CONFIGS: Dict[str, Dict[str, Any]] = {261 "task1": {262 "description": (263 "Waste Cleanup: terminate all idle EBS volumes and S3 buckets "264 "(cpu_utilization == 0, idle_hours > 0). "265 "Do NOT terminate critical resources or break dependency chains."266 ),267 "resources_fn": _task1_resources,268 "budget_per_hour": 1.00,269 "max_steps": 12,270 },271 "task2": {272 "description": (273 "Resource Optimization: resize over-provisioned EC2 instances "274 "(cpu_utilization < 20%) down by at least one tier. "275 "Check SLA constraints before resizing — some instances have tight "276 "cpu caps (sla_max_cpu). Resizes take 2 steps to stabilise."277 ),278 "resources_fn": _task2_resources,279 "budget_per_hour": 0.80,280 "max_steps": 16,281 },282 "task3": {283 "description": (284 "Strategic Planning: use simulate to reason about actions, then "285 "terminate idle waste and reserve always-on instances to bring "286 "total hourly spend below $1.00/hr. Preserve the 3-tier dependency "287 "chain (frontend -> backend -> db) and respect SLA caps. "288 "Reservations are permanent commitments."289 ),290 "resources_fn": _task3_resources,291 "budget_per_hour": 1.00,292 "max_steps": 20,293 },294}295 296 297# ---------------------------------------------------------------------------298# Environment299# ---------------------------------------------------------------------------300 301class CloudFinOpsEnvironment(Environment):302 """303 Cloud FinOps Optimizer — production-grade OpenEnv environment.304 305 Implements the OpenEnv Environment interface:306 reset(**kwargs) -> FinOpsObservation307 step(action) -> FinOpsObservation308 state -> FinOpsState (property)309 310 Additional method:311 grade() -> Dict[str, Any] — deterministic episode score312 """313 314 SUPPORTS_CONCURRENT_SESSIONS = True315 316 def __init__(self) -> None:317 self._state = FinOpsState()318 self._resources: List[CloudResource] = []319 320 # ------------------------------------------------------------------321 # OpenEnv interface322 # ------------------------------------------------------------------323 324 def reset(self, seed: Optional[int] = None, episode_id: Optional[str] = None, **kwargs) -> FinOpsObservation:325 # Extract task_id from kwargs if provided, otherwise default to task1326 task_id = kwargs.get("task_id", "task1")327 if task_id not in TASK_CONFIGS:328 raise ValueError(f"Unknown task_id '{task_id}'. Choose from: {list(TASK_CONFIGS)}")329 330 cfg = TASK_CONFIGS[task_id]331 self._resources = cfg["resources_fn"]()332 total_cost = self._active_cost()333 334 self._state = FinOpsState(335 episode_id=episode_id or str(uuid.uuid4()),336 step_count=0,337 task_id=task_id,338 total_cost_per_hour=total_cost,339 initial_cost_per_hour=total_cost,340 budget_per_hour=cfg["budget_per_hour"],341 done=False,342 )343 return self._build_observation(reward=None, done=False)344 345 def step(self, action: FinOpsAction, **kwargs) -> FinOpsObservation:346 if not self._state.task_id:347 raise RuntimeError("Environment not initialized. Call reset() first.")348 if self._state.done:349 return self._build_observation(350 reward=0.0, done=True,351 info={"message": "Episode already finished."},352 last_action_error="Episode already finished.",353 )354 355 # Upgrade 4: log agent reasoning356 if action.reasoning:357 self._state.reasoning_log.append(358 f"step{self._state.step_count}: {action.reasoning.strip()}"359 )360 361 # Upgrade 5: simulate mode — no state mutation362 if action.action_type == "simulate":363 sim_result = self._simulate(action.simulate_action or {})364 self._state.step_count += 1365 return self._build_observation(366 reward=0.0, done=False,367 simulate_result=sim_result,368 )369 370 # Upgrade 3: advance existing cooldown timers BEFORE applying the new action371 # so that a freshly-set cooldown is not decremented on the same step372 self._tick_cooldowns()373 374 reward, info = self._apply_action(action)375 self._state.step_count += 1376 self._state.total_cost_per_hour = self._active_cost()377 378 cfg = TASK_CONFIGS[self._state.task_id]379 done = self._state.step_count >= cfg["max_steps"]380 self._state.done = done381 382 error_str = info.get("error") if isinstance(info, dict) else None383 return self._build_observation(384 reward=round(reward, 4),385 done=done,386 info=info,387 last_action_error=error_str,388 )389 390 @property391 def state(self) -> FinOpsState:392 return self._state393 394 # ------------------------------------------------------------------395 # Deterministic graders (0.0 – 1.0)396 # ------------------------------------------------------------------397 398 def grade(self) -> Dict[str, Any]:399 task_id = self._state.task_id400 if not task_id:401 return {"score": 0.01, "reason": "No active episode."}402 403 if task_id == "task1":404 result = self._grade_task1()405 elif task_id == "task2":406 result = self._grade_task2()407 elif task_id == "task3":408 result = self._grade_task3()409 else:410 return {"score": 0.01, "reason": f"Unknown task {task_id}"}411 412 # Calculate bonus score413 expl_score = self._score_reasoning()414 result["explainability_score"] = expl_score415 416 # Phase 2 Strict Constraint: Score must be strictly between 0 and 1 (not 0 or 1).417 # We clamp the FINAL combined score here.418 final_score = result["score"] + (expl_score * 0.05)419 result["score"] = round(max(0.01, min(0.99, final_score)), 4)420 421 return result422 423 def _score_reasoning(self) -> float:424 """425 Advanced scoring logic for agent reasoning quality.426 Recognizes keyword coverage, logical structure, and contextual awareness.427 """428 log = self._state.reasoning_log429 if not log:430 return 0.1 # Base score for valid but empty history431 432 import re433 recent = log[-5:] # look at last 5 decisions434 435 keywords = {436 "cost": 0.1, "waste": 0.1, "idle": 0.1,437 "sla": 0.15, "breach": 0.15, "violation": 0.15,438 "dependency": 0.15, "cascade": 0.15, "chain": 0.15,439 "simulate": 0.15, "projected": 0.1, "safe": 0.1440 }441 442 hits = 0.0443 for text in recent:444 text_lower = text.lower()445 # 1. Keyword coverage446 for kw, val in keywords.items():447 if kw in text_lower:448 hits += val449 450 # 2. Justification markers (e.g., "because", "due to")451 if re.search(r"(because|due|since|result|reason|why)", text_lower):452 hits += 0.2453 454 # 3. Contextual markers (e.g., resource IDs like 'ec2-001')455 if re.search(r"[a-z0-9]+-[a-z0-9]+", text_lower):456 hits += 0.2457 458 # 4. Strategy markers459 if "simulate" in text_lower and ("safe" in text_lower or "advice" in text_lower):460 hits += 0.2461 462 # Average hits per logged reasoning463 raw_score = 0.1 + (hits / len(recent))464 return round(max(0.1, min(0.99, raw_score)), 4)465 466 def _grade_task1(self) -> Dict[str, Any]:467 original = _task1_resources()468 waste_ids = {r.id for r in original if not r.critical and r.cpu_utilization == 0.0 and r.idle_hours > 0}469 critical_ids = {r.id for r in original if r.critical}470 by_id = {r.id: r for r in self._resources}471 472 correctly_terminated = sum(1 for rid in waste_ids if by_id[rid].status == ResourceStatus.TERMINATED)473 critical_terminated = sum(1 for rid in critical_ids if by_id[rid].status == ResourceStatus.TERMINATED)474 sla_violations = sum(1 for r in self._resources if r.sla_status == SLAStatus.VIOLATED)475 476 base = correctly_terminated / len(waste_ids) if waste_ids else 1.0477 score = base - 0.25 * critical_terminated - 0.05 * sla_violations478 479 # Phase 2 Strict Constraint: strictly between 0 and 1480 score = max(0.01, min(0.99, score))481 482 return {483 "score": round(score, 4),484 "waste_resources": len(waste_ids),485 "correctly_terminated": correctly_terminated,486 "critical_terminated": critical_terminated,487 "sla_violations": sla_violations,488 }489 490 def _grade_task2(self) -> Dict[str, Any]:491 original = _task2_resources()492 overprovisioned = {r.id for r in original if r.type == ResourceType.EC2 and r.cpu_utilization < 20.0}493 critical_ids = {r.id for r in original if r.critical}494 orig_by_id = {r.id: r for r in original}495 by_id = {r.id: r for r in self._resources}496 497 correctly_resized = sum(498 1 for rid in overprovisioned499 if by_id[rid].status != ResourceStatus.TERMINATED500 and by_id[rid].instance_size is not None501 and orig_by_id[rid].instance_size is not None502 and _size_index(by_id[rid].instance_size) < _size_index(orig_by_id[rid].instance_size)503 and by_id[rid].sla_status != SLAStatus.VIOLATED # SLA must hold504 )505 critical_terminated = sum(1 for rid in critical_ids if by_id[rid].status == ResourceStatus.TERMINATED)506 sla_violations = sum(1 for r in self._resources if r.sla_status == SLAStatus.VIOLATED)507 508 base = correctly_resized / len(overprovisioned) if overprovisioned else 1.0509 score = base - 0.25 * critical_terminated - 0.05 * sla_violations510 511 # Phase 2 Strict Constraint: strictly between 0 and 1512 score = max(0.01, min(0.99, score))513 514 return {515 "score": round(score, 4),516 "overprovisioned_resources": len(overprovisioned),517 "correctly_resized": correctly_resized,518 "critical_terminated": critical_terminated,519 "sla_violations": sla_violations,520 }521 522 def _grade_task3(self) -> Dict[str, Any]:523 original = _task3_resources()524 critical_ids = {r.id for r in original if r.critical}525 by_id = {r.id: r for r in self._resources}526 527 current_cost = self._active_cost()528 budget = self._state.budget_per_hour529 initial = self._state.initial_cost_per_hour530 critical_terminated = sum(1 for rid in critical_ids if by_id[rid].status == ResourceStatus.TERMINATED)531 sla_violations = sum(1 for r in self._resources if r.sla_status == SLAStatus.VIOLATED)532 533 if current_cost <= budget:534 base = 1.0535 else:536 needed = initial - budget537 base = min(1.0, max(0.0, (initial - current_cost) / needed)) if needed > 0 else 1.0538 539 score = base - 0.30 * critical_terminated - 0.05 * sla_violations540 541 # Phase 2 Strict Constraint: strictly between 0 and 1542 score = max(0.01, min(0.99, score))543 544 return {545 "score": round(score, 4),546 "initial_cost_per_hour": round(initial, 4),547 "current_cost_per_hour": round(current_cost, 4),548 "budget_per_hour": budget,549 "under_budget": current_cost <= budget,550 "critical_terminated": critical_terminated,551 "sla_violations": sla_violations,552 }553 554 # ------------------------------------------------------------------555 # Upgrade 5: Simulate (no state mutation)556 # ------------------------------------------------------------------557 558 def _simulate(self, proposed: Dict[str, Any]) -> SimulateResult:559 """560 Project the outcome of a proposed action without changing any state.561 Returns a SimulateResult the agent can inspect before committing.562 """563 action_type = proposed.get("action_type", "noop")564 resource_id = proposed.get("resource_id")565 target_size = proposed.get("target_size")566 567 projected_cost = self._active_cost()568 projected_reward = 0.0569 sla_violations: List[str] = []570 cascade_risks: List[str] = []571 safe = True572 recommendation = "Action appears safe."573 574 r = self._find(resource_id) if resource_id else None575 576 if action_type == "terminate" and r is not None:577 # Always compute cascade risks (useful even for critical resources)578 for other in self._resources:579 if r.id in (other.dependency_ids or []) and other.status == ResourceStatus.RUNNING:580 cascade_risks.append(other.id)581 582 if r.critical:583 projected_reward = -1.0584 safe = False585 cascade_note = f" Also cascades to: {cascade_risks}." if cascade_risks else ""586 recommendation = (587 f"UNSAFE: '{r.id}' is critical. Terminating it causes a -1.0 penalty.{cascade_note}"588 )589 elif cascade_risks:590 projected_cost -= r.cost_per_hour591 safe = False592 recommendation = (593 f"WARNING: terminating '{r.id}' will cascade to: {cascade_risks}. "594 "Penalty -0.20 per affected resource."595 )596 else:597 projected_cost -= r.cost_per_hour598 idle_bonus = 0.10 if r.cpu_utilization == 0.0 and r.idle_hours > 100 else 0.0599 projected_reward = min(0.5, r.cost_per_hour * 2.0) + idle_bonus600 recommendation = (601 f"SAFE: terminating '{r.id}' saves ${r.cost_per_hour:.4f}/hr. "602 f"Projected reward: {projected_reward:+.4f}."603 )604 605 elif action_type == "resize" and r is not None and target_size is not None:606 try:607 new_size = InstanceSize(target_size)608 proj_cpu = _projected_cpu_after_resize(r, new_size)609 cost_saving = r.cost_per_hour - INSTANCE_COST[new_size]610 projected_cost -= cost_saving611 if proj_cpu > r.sla_max_cpu:612 sla_violations.append(r.id)613 safe = False614 recommendation = (615 f"UNSAFE: resizing '{r.id}' to {new_size} projects CPU at "616 f"{proj_cpu:.1f}% (SLA cap {r.sla_max_cpu:.1f}%). "617 f"Consider a less aggressive resize."618 )619 else:620 projected_reward = min(0.5, cost_saving * 3.0)621 if r.cpu_utilization < 20.0:622 projected_reward += 0.10623 recommendation = (624 f"SAFE: resize '{r.id}' to {new_size} saves ${cost_saving:.4f}/hr. "625 f"Projected CPU: {proj_cpu:.1f}% (cap {r.sla_max_cpu:.1f}%). "626 f"Reward: {projected_reward:+.4f}. "627 f"Note: cooldown {RESIZE_COOLDOWN_STEPS} steps."628 )629 except ValueError:630 safe = False631 recommendation = f"INVALID: '{target_size}' is not a valid instance size."632 633 elif action_type == "reserve" and r is not None:634 if r.reserved:635 projected_reward = -0.02636 recommendation = f"'{r.id}' is already reserved. No effect."637 else:638 saving = r.cost_per_hour * RESERVE_DISCOUNT639 projected_cost -= saving640 projected_reward = min(0.5, saving * 2.5)641 if r.cpu_utilization > 50.0 and r.cost_per_hour > 0.10:642 projected_reward += 0.10643 recommendation = (644 f"SAFE: reserving '{r.id}' saves ${saving:.4f}/hr (40% discount). "645 f"Reward: {projected_reward:+.4f}. "646 f"NOTE: reservation is a permanent commitment."647 )648 649 elif action_type == "noop":650 recommendation = "Noop applies a -0.02 penalty. Only use if no better action exists."651 652 return SimulateResult(653 proposed_action=proposed,654 projected_cost_per_hour=round(projected_cost, 6),655 projected_budget_remaining=round(self._state.budget_per_hour - projected_cost, 6),656 projected_reward=round(projected_reward, 4),657 projected_sla_violations=sla_violations,658 cascading_risks=cascade_risks,659 recommendation=recommendation,660 safe_to_apply=safe,661 )662 663 # ------------------------------------------------------------------664 # Upgrade 3: Cooldown ticker665 # ------------------------------------------------------------------666 667 def _tick_cooldowns(self) -> None:668 """669 Advance resize cooldown timers; mark SLA status accordingly.670 Only updates sla_status to AT_RISK if the resource is not already VIOLATED.671 Expired cooldowns trigger a full SLA re-evaluation.672 """673 for r in self._resources:674 if r.resize_cooldown_steps > 0:675 r.resize_cooldown_steps -= 1676 if r.resize_cooldown_steps > 0:677 # Still in cooldown — mark at-risk only if not already violated678 if r.sla_status != SLAStatus.VIOLATED:679 r.sla_status = SLAStatus.AT_RISK680 else:681 # Cooldown expired — re-evaluate SLA from actual cpu_utilization682 self._update_sla_status(r)683 684 # ------------------------------------------------------------------685 # Internal helpers686 # ------------------------------------------------------------------687 688 def _active_cost(self) -> float:689 return round(690 sum(r.cost_per_hour for r in self._resources if r.status != ResourceStatus.TERMINATED),691 6,692 )693 694 def _find(self, resource_id: Optional[str]) -> Optional[CloudResource]:695 if resource_id is None:696 return None697 for r in self._resources:698 if r.id == resource_id:699 return r700 return None701 702 def _update_sla_status(self, r: CloudResource) -> None:703 if r.cpu_utilization > r.sla_max_cpu:704 r.sla_status = SLAStatus.VIOLATED705 if r.id not in self._state.sla_violation_history:706 self._state.sla_violation_history.append(r.id)707 elif r.cpu_utilization > r.sla_max_cpu * 0.85:708 r.sla_status = SLAStatus.AT_RISK709 else:710 r.sla_status = SLAStatus.OK711 712 def _build_dependency_graph(self) -> Tuple[Dict[str, List[str]], Dict[str, List[str]]]:713 """714 Returns:715 dep_graph : resource_id -> list of its dependency_ids (what it needs)716 cascade_risks : resource_id -> list of resources that depend on it717 """718 dep_graph: Dict[str, List[str]] = {}719 cascade_risks: Dict[str, List[str]] = {}720 active = [r for r in self._resources if r.status != ResourceStatus.TERMINATED]721 for r in active:722 dep_graph[r.id] = [d for d in r.dependency_ids723 if any(x.id == d and x.status != ResourceStatus.TERMINATED724 for x in active)]725 for r in active:726 for dep_id in r.dependency_ids:727 cascade_risks.setdefault(dep_id, [])728 if r.id not in cascade_risks[dep_id]:729 cascade_risks[dep_id].append(r.id)730 return dep_graph, cascade_risks731 732 def _active_sla_violations(self) -> List[str]:733 return [r.id for r in self._resources if r.sla_status == SLAStatus.VIOLATED]734 735 def _build_observation(736 self,737 reward: Optional[float],738 done: bool,739 info: Optional[Dict[str, Any]] = None,740 last_action_error: Optional[str] = None,741 simulate_result: Optional[SimulateResult] = None,742 ) -> FinOpsObservation:743 cfg = TASK_CONFIGS.get(self._state.task_id, {})744 cost = self._active_cost()745 desc = cfg.get("description", "")746 dep_graph, cascade_risks = self._build_dependency_graph()747 748 info = info or {749 "episode_id": self._state.episode_id,750 "terminated_count": len(self._state.terminated_ids),751 "reserved_count": len(self._state.reserved_ids),752 "resize_count": len(self._state.resize_history),753 "sla_violations": self._state.sla_violation_history,754 }755 info["finops_metrics"] = {756 "active_waste_count": sum(1 for r in self._resources if not r.critical and r.cpu_utilization == 0.0 and r.status != ResourceStatus.TERMINATED),757 "at_risk_count": sum(1 for r in self._resources if r.sla_status == SLAStatus.AT_RISK),758 "violation_count": sum(1 for r in self._resources if r.sla_status == SLAStatus.VIOLATED),759 "total_savings_potential_hr": sum(r.cost_per_hour for r in self._resources if not r.critical and r.cpu_utilization < 20.0 and r.status != ResourceStatus.TERMINATED),760 }761 762 return FinOpsObservation(763 done=done,764 reward=reward,765 resources=self._resources,766 total_cost_per_hour=cost,767 budget_per_hour=self._state.budget_per_hour,768 budget_remaining=round(self._state.budget_per_hour - cost, 6),769 task_id=self._state.task_id,770 task_description=desc,771 goal=(772 f"[{self._state.task_id}] Reduce cloud costs below "773 f"${self._state.budget_per_hour:.2f}/hr. "774 f"{desc.split(chr(46))[0]}."775 ),776 last_action_error=last_action_error,777 dependency_graph=dep_graph,778 cascading_risks=cascade_risks,779 sla_violations=self._active_sla_violations(),780 simulate_result=simulate_result,781 step_count=self._state.step_count,782 max_steps=cfg.get("max_steps", 20),783 info=info or {784 "episode_id": self._state.episode_id,785 "terminated_count": len(self._state.terminated_ids),786 "reserved_count": len(self._state.reserved_ids),787 "resize_count": len(self._state.resize_history),788 "sla_violations": self._state.sla_violation_history,789 },790 )791 792 # ------------------------------------------------------------------793 # Action dispatch794 # ------------------------------------------------------------------795 796 def _apply_action(self, action: FinOpsAction) -> Tuple[float, Dict[str, Any]]:797 t = action.action_type798 if t == "noop":799 return -0.02, {"explanation": "No action taken."}800 if t == "terminate":801 return self._terminate(action.resource_id)802 if t == "resize":803 return self._resize(action.resource_id, action.target_size)804 if t == "reserve":805 return self._reserve(action.resource_id)806 return -0.05, {"error": f"Unknown action_type '{t}'."}807 808 def _terminate(self, resource_id: Optional[str]) -> Tuple[float, Dict[str, Any]]:809 r = self._find(resource_id)810 if r is None:811 return -0.05, {"error": f"Resource '{resource_id}' not found."}812 if r.status == ResourceStatus.TERMINATED:813 return -0.03, {"explanation": "Already terminated."}814 if r.critical:815 r.status = ResourceStatus.TERMINATED816 self._state.terminated_ids.append(r.id)817 return -1.0, {"explanation": f"CRITICAL VIOLATION: terminated '{r.id}'."}818 819 # Upgrade 1: cascade penalty820 dep_penalty = 0.0821 affected = []822 for other in self._resources:823 if r.id in other.dependency_ids and other.status == ResourceStatus.RUNNING:824 dep_penalty -= 0.20825 affected.append(other.id)826 827 old_cost = r.cost_per_hour828 r.status = ResourceStatus.TERMINATED829 self._state.terminated_ids.append(r.id)830 831 cost_reward = min(0.5, old_cost * 2.0)832 idle_bonus = 0.10 if r.cpu_utilization == 0.0 and r.idle_hours > 100 else 0.0833 waste_penalty = -0.15 if r.cpu_utilization > 30.0 else 0.0834 total = max(-1.0, min(1.0, cost_reward + idle_bonus + waste_penalty + dep_penalty))835 info = {"explanation": f"Terminated '{r.id}'."}836 if affected:837 info["cascade_affected"] = affected838 return total, info839 840 def _resize(self, resource_id: Optional[str], target_size: Optional[str]) -> Tuple[float, Dict[str, Any]]:841 r = self._find(resource_id)842 if r is None:843 return -0.05, {"error": f"Resource '{resource_id}' not found."}844 if r.status == ResourceStatus.TERMINATED:845 return -0.03, {"explanation": "Cannot resize a terminated resource."}846 if r.type != ResourceType.EC2 or r.instance_size is None:847 return -0.05, {"explanation": "Resize only valid for EC2 instances."}848 try:849 new_size = InstanceSize(target_size)850 except ValueError:851 return -0.05, {"error": f"Invalid target_size '{target_size}'."}852 if _size_index(new_size) >= _size_index(r.instance_size):853 return -0.05, {"explanation": "target_size must be smaller than current size."}854 855 old_cost = r.cost_per_hour856 old_size = r.instance_size # capture BEFORE mutation for history857 old_cpu = r.cpu_utilization # capture BEFORE mutation for reward calc858 proj_cpu = _projected_cpu_after_resize(r, new_size)859 860 # Upgrade 2: SLA check against projected CPU861 sla_penalty = 0.0862 if proj_cpu > r.sla_max_cpu:863 r.sla_status = SLAStatus.VIOLATED864 sla_penalty = -0.20865 if r.id not in self._state.sla_violation_history:866 self._state.sla_violation_history.append(r.id)867 868 # Upgrade 3: apply mutation — cooldown, new size, projected cpu869 r.instance_size = new_size870 r.cost_per_hour = INSTANCE_COST[new_size]871 r.cpu_utilization = proj_cpu872 r.resize_cooldown_steps = RESIZE_COOLDOWN_STEPS873 if r.sla_status != SLAStatus.VIOLATED:874 r.sla_status = SLAStatus.AT_RISK # uncertain until cooldown expires875 # Record actual old_size -> new_size (old_size captured before mutation)876 self._state.resize_history[r.id] = f"{old_size.value}->{new_size.value}"877 878 cost_reward = min(0.5, (old_cost - r.cost_per_hour) * 3.0)879 overprovisioned_bonus = 0.10 if old_cpu < 20.0 else 0.0 # use pre-resize cpu880 risk_penalty = -0.10 if proj_cpu > 70.0 else 0.0 # use post-resize cpu881 total = max(-1.0, min(1.0, cost_reward + overprovisioned_bonus + risk_penalty + sla_penalty))882 return total, {883 "explanation": f"Resized '{r.id}' to {new_size}. Projected CPU: {proj_cpu:.1f}%.",884 "cooldown_steps": RESIZE_COOLDOWN_STEPS,885 "sla_status": r.sla_status.value,886 }887 888 def _reserve(self, resource_id: Optional[str]) -> Tuple[float, Dict[str, Any]]:889 r = self._find(resource_id)890 if r is None:891 return -0.05, {"error": f"Resource '{resource_id}' not found."}892 if r.status == ResourceStatus.TERMINATED:893 return -0.03, {"explanation": "Cannot reserve a terminated resource."}894 if r.reserved:895 return -0.02, {"explanation": "Resource already reserved."}896 897 old_cost = r.cost_per_hour898 r.reserved = True899 r.reservation_committed = True # Upgrade 3: permanent900 r.cost_per_hour = round(old_cost * (1.0 - RESERVE_DISCOUNT), 6)901 self._state.reserved_ids.append(r.id)902 903 cost_reward = min(0.5, (old_cost - r.cost_per_hour) * 2.5)904 strategy_bonus = 0.10 if r.cpu_utilization > 50.0 and old_cost > 0.10 else 0.0905 total = max(-1.0, min(1.0, cost_reward + strategy_bonus))906 return total, {907 "explanation": f"Reserved '{r.id}' at {int(RESERVE_DISCOUNT*100)}% discount. Permanent commitment.",908 }909