# -*- coding: utf-8 -*- """Shared state and helpers used by all routers. Centralizes session registries, telemetry counters, leaderboard, and Prometheus metric helpers so that routers can import them without circular dependencies back to app.py. """ from __future__ import annotations import logging import os import re import secrets import time from typing import Any, Dict, List _log = logging.getLogger("irt.api") # --------------------------------------------------------------------------- # Session-isolated environment registry # --------------------------------------------------------------------------- _SESSION_REGISTRY: Dict[str, Any] = {} _SESSION_TIMESTAMPS: Dict[str, float] = {} _MAX_SESSIONS = 256 _SESSION_TTL = int(os.environ.get("SESSION_TTL_SECONDS", 3600)) # SENTINEL session registry (separate from IRT) _SENTINEL_REGISTRY: Dict[str, Any] = {} _SENTINEL_TIMESTAMPS: Dict[str, float] = {} # --------------------------------------------------------------------------- # Telemetry counters (in-process; reset on restart) # --------------------------------------------------------------------------- _TELEMETRY: Dict[str, int] = { "sessions_created": 0, "sessions_evicted_fifo": 0, "sessions_expired_ttl": 0, "episodes_total": 0, "steps_total": 0, "grader_calls": 0, "baseline_runs": 0, "errors_total": 0, "ws_connections_total": 0, "sentinel_sessions_created": 0, "sentinel_episodes_total": 0, "sentinel_steps_total": 0, "sentinel_grader_calls": 0, } # Active WebSocket connections (single-process; decremented on disconnect) WS_ACTIVE_CONNECTIONS: int = 0 # --------------------------------------------------------------------------- # In-memory leaderboard (top-10 scores per task) # --------------------------------------------------------------------------- _LEADERBOARD: Dict[str, List[Dict[str, Any]]] = { "severity_classification": [], "root_cause_analysis": [], "full_incident_management": [], "basic_oversight": [], "fleet_monitoring_conflict": [], "adversarial_worker": [], "multi_crisis_command": [], } _LEADERBOARD_SIZE = 10 # --------------------------------------------------------------------------- # Session helpers # --------------------------------------------------------------------------- def get_or_create_session(session_id: str | None): """Return (session_id, env). Creates a new session if id is None or unknown.""" from src.environment import IncidentResponseEnv if session_id and session_id in _SESSION_REGISTRY: return session_id, _SESSION_REGISTRY[session_id] # New session - evict if at capacity if len(_SESSION_REGISTRY) >= _MAX_SESSIONS: oldest = next(iter(_SESSION_REGISTRY)) del _SESSION_REGISTRY[oldest] _SESSION_TIMESTAMPS.pop(oldest, None) _TELEMETRY["sessions_evicted_fifo"] += 1 _log.info("session evicted (FIFO): %s", oldest) new_id = session_id or secrets.token_hex(16) _SESSION_REGISTRY[new_id] = IncidentResponseEnv() _SESSION_TIMESTAMPS[new_id] = time.time() _TELEMETRY["sessions_created"] += 1 return new_id, _SESSION_REGISTRY[new_id] def get_or_create_sentinel_session(session_id: str | None): """Return (session_id, sentinel_env). Creates a new SENTINEL session if id is None or unknown.""" from sentinel.environment import SentinelEnv if session_id and session_id in _SENTINEL_REGISTRY: return session_id, _SENTINEL_REGISTRY[session_id] # New session - evict if at capacity if len(_SENTINEL_REGISTRY) >= _MAX_SESSIONS: oldest = next(iter(_SENTINEL_REGISTRY)) del _SENTINEL_REGISTRY[oldest] _SENTINEL_TIMESTAMPS.pop(oldest, None) _TELEMETRY["sessions_evicted_fifo"] += 1 _log.info("sentinel session evicted (FIFO): %s", oldest) new_id = session_id or secrets.token_hex(16) _SENTINEL_REGISTRY[new_id] = SentinelEnv() _SENTINEL_TIMESTAMPS[new_id] = time.time() _TELEMETRY["sentinel_sessions_created"] += 1 return new_id, _SENTINEL_REGISTRY[new_id] def purge_expired_sessions() -> int: """Remove sessions older than SESSION_TTL. Returns number purged.""" cutoff = time.time() - _SESSION_TTL stale = [sid for sid, ts in _SESSION_TIMESTAMPS.items() if ts < cutoff] stale_sentinel = [sid for sid, ts in _SENTINEL_TIMESTAMPS.items() if ts < cutoff] for sid in stale: _SESSION_REGISTRY.pop(sid, None) _SESSION_TIMESTAMPS.pop(sid, None) _TELEMETRY["sessions_expired_ttl"] += 1 for sid in stale_sentinel: _SENTINEL_REGISTRY.pop(sid, None) _SENTINEL_TIMESTAMPS.pop(sid, None) _TELEMETRY["sessions_expired_ttl"] += 1 total_purged = len(stale) + len(stale_sentinel) if total_purged: _log.info("purged %d stale session(s) (%d IRT, %d SENTINEL)", total_purged, len(stale), len(stale_sentinel)) return total_purged def record_leaderboard(task_id: str, score: float, steps: int) -> None: """Insert a completed episode score into the in-memory leaderboard.""" board = _LEADERBOARD.get(task_id) if board is None: return board.append({"score": score, "steps": steps, "ts": round(time.time())}) board.sort(key=lambda e: (-e["score"], e["steps"])) del board[_LEADERBOARD_SIZE:] # keep top-N # --------------------------------------------------------------------------- # Prometheus metric helpers # --------------------------------------------------------------------------- # (prom_metric_name, ServiceMetrics field, HELP text) _PROM_CORE_FIELDS: List[tuple] = [ ("irt_cpu_percent", "cpu_percent", "CPU utilisation percent"), ("irt_memory_percent", "memory_percent", "Memory utilisation percent"), ("irt_request_rate", "request_rate", "Requests per second"), ("irt_error_rate", "error_rate", "HTTP error rate fraction 0.0-1.0"), ("irt_latency_p50_ms", "latency_p50_ms", "P50 response latency milliseconds"), ("irt_latency_p99_ms", "latency_p99_ms", "P99 response latency milliseconds"), ] def scenario_live_to_prom_text( live: Dict[str, Any], scenario_id: str, incident_id: str, step: int, ) -> str: """Serialize live scenario metrics to Prometheus text exposition format.""" lines: List[str] = [ f'# HELP irt_scenario_step Current episode step number', f'# TYPE irt_scenario_step gauge', f'irt_scenario_step{{scenario="{scenario_id}",incident="{incident_id}"}} {step}', ] for prom_name, field, help_text in _PROM_CORE_FIELDS: lines += [ f"# HELP {prom_name} {help_text}", f"# TYPE {prom_name} gauge", ] for svc, m in live.items(): val = getattr(m, field, 0.0) lines.append( f'{prom_name}{{service="{svc}",scenario="{scenario_id}",incident="{incident_id}"}} {val}' ) # Custom metrics (e.g. connection_pool_used, heap_mb, ...) all_custom: Dict[str, str] = {} # prom_name -> raw key for m in live.values(): for raw_key in (m.custom or {}): prom_key = "irt_custom_" + re.sub(r"[^a-zA-Z0-9_]", "_", raw_key) all_custom[prom_key] = raw_key for prom_key in sorted(all_custom): raw_key = all_custom[prom_key] lines += [ f"# HELP {prom_key} Custom scenario metric: {raw_key}", f"# TYPE {prom_key} gauge", ] for svc, m in live.items(): val = (m.custom or {}).get(raw_key) if val is not None: lines.append( f'{prom_key}{{service="{svc}",scenario="{scenario_id}",incident="{incident_id}"}} {val}' ) return "\n".join(lines) + "\n" _PROM_SELECTOR_RE = re.compile( r"^(?P[a-zA-Z_:][a-zA-Z0-9_:]*)?(?:\{(?P[^}]*)\})?$" ) _PROM_LABEL_RE = re.compile(r'(\w+)\s*=\s*"([^"]*)"') def parse_prom_selector(query: str) -> tuple[str, Dict[str, str]]: """Parse a simple PromQL instant selector into (metric_name, label_filters).""" m = _PROM_SELECTOR_RE.match(query.strip()) if not m: return query.strip(), {} name = m.group("name") or "" label_str = m.group("labels") or "" filters: Dict[str, str] = { lm.group(1): lm.group(2) for lm in _PROM_LABEL_RE.finditer(label_str) } return name, filters def build_prom_vector( live: Dict[str, Any], metric_name: str, label_filters: Dict[str, str], scenario_id: str, incident_id: str, ) -> List[Dict[str, Any]]: """Build a Prometheus instant-query vector result list.""" ts = round(time.time(), 3) # Normalise: auto-prefix irt_ when caller omits it if metric_name and not metric_name.startswith("irt_"): metric_name = f"irt_{metric_name}" field_map = {pn: fn for pn, fn, _ in _PROM_CORE_FIELDS} candidates = [metric_name] if metric_name else [pn for pn, _, _ in _PROM_CORE_FIELDS] results: List[Dict[str, Any]] = [] for prom_name in candidates: field = field_map.get(prom_name) for svc, m in live.items(): if "service" in label_filters and label_filters["service"] != svc: continue if "scenario" in label_filters and label_filters["scenario"] != scenario_id: continue if field is not None: val = getattr(m, field, 0.0) elif prom_name.startswith("irt_custom_"): raw_key = prom_name[len("irt_custom_"):] val = (m.custom or {}).get(raw_key) if val is None: continue else: continue results.append({ "metric": { "__name__": prom_name, "service": svc, "scenario": scenario_id, "incident": incident_id, }, "value": [ts, str(val)], }) return results def build_prom_matrix( history: Dict[str, Any], metric_name: str, label_filters: Dict[str, str], scenario_id: str, incident_id: str, ) -> List[Dict[str, Any]]: """Build a Prometheus range-query matrix result from ring-buffer history. ``history`` is the dict returned by ``env.metric_history(start, end)``: {service_name: [(ts, ServiceMetrics), ...], ...} Returns the standard Prometheus matrix result shape: [{"metric": {...labels}, "values": [[ts, "value"], ...]}, ...] """ if metric_name and not metric_name.startswith("irt_"): metric_name = f"irt_{metric_name}" field_map = {pn: fn for pn, fn, _ in _PROM_CORE_FIELDS} candidates = [metric_name] if metric_name else [pn for pn, _, _ in _PROM_CORE_FIELDS] # Build one result stream per (prom_name, service) streams: Dict[tuple, List] = {} # (prom_name, svc) -> [[ts, "val"],...] for svc, samples in history.items(): if "service" in label_filters and label_filters["service"] != svc: continue if "scenario" in label_filters and label_filters["scenario"] != scenario_id: continue for prom_name in candidates: field = field_map.get(prom_name) for ts, m in samples: if field is not None: val = getattr(m, field, 0.0) elif prom_name.startswith("irt_custom_"): raw_key = prom_name[len("irt_custom_"):] val = (m.custom or {}).get(raw_key) if val is None: continue else: continue key = (prom_name, svc) if key not in streams: streams[key] = [] streams[key].append([round(ts, 3), str(val)]) results: List[Dict[str, Any]] = [] for (prom_name, svc), values in streams.items(): results.append({ "metric": { "__name__": prom_name, "service": svc, "scenario": scenario_id, "incident": incident_id, }, "values": sorted(values, key=lambda x: x[0]), }) return results