Spaces:
Runtime error
Runtime error
File size: 2,526 Bytes
205f6c7 13558c9 205f6c7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | # Task: Critical — SLA-breach scenario with controlled randomness and realism.
# Introduces noisy logs, metric variance, and slight ambiguity while preserving solvability.
from __future__ import annotations
from typing import Dict, Any, Tuple
import random
from env.environment import DebugEnv
class CriticalEnv(DebugEnv):
"""
High-stakes incident:
- Root cause: memory_leak (restart_api → restart_db)
- Noisy + misleading logs
- Randomized metrics per episode
- SLA penalties + time pressure
- Designed for score variance across runs
"""
SLA_LATENCY_THRESHOLD = 250 # ms
SLA_PENALTY = 80.0
TIME_PRESSURE_START = 8
TIME_PRESSURE_PENALTY = 30.0
def reset(self) -> Dict[str, Any]:
state = super().reset()
self.state_data["root_cause"] = "memory_leak"
self.state_data["fix_sequence"] = ["restart_api", "restart_db"]
self.state_data["services"]["api"] = "degraded"
self.state_data["services"]["db"] = "degraded"
base_logs = [
"memory usage increasing",
"OOM warning",
"heap allocation failure",
]
noise_logs = [
"network latency spike (transient)",
"disk almost full: /var/log 94%",
"temporary service restart: metrics-collector",
"cache eviction rate elevated",
"connection pool retry",
"upstream request timeout",
]
# Randomly inject noise
selected_noise = random.sample(noise_logs, k=random.randint(2, 4))
logs = base_logs + selected_noise
random.shuffle(logs)
self.state_data["logs"] = logs
self.state_data["metrics"]["latency"] = random.randint(350, 500)
self.state_data["metrics"]["error_rate"] = round(random.uniform(0.6, 0.9), 2)
self.state_data["metrics"]["cpu"] = random.randint(85, 98)
self._prev_latency = self.state_data["metrics"]["latency"]
return self._obs()
def step(self, action: str) -> Tuple[Dict[str, Any], float, bool, Dict[str, Any]]:
obs, reward, done, info = super().step(action)
if info["latency"] > self.SLA_LATENCY_THRESHOLD:
reward -= self.SLA_PENALTY
if self.t > self.TIME_PRESSURE_START and not info["resolved"]:
reward -= self.TIME_PRESSURE_PENALTY
reward += random.uniform(-3, 3)
return obs, reward, done, info
def create_env() -> CriticalEnv:
return CriticalEnv(max_steps=10) |