""" Ticket generation scenarios for SupportOps Arena. Defines root causes and generates tickets for easy, medium, and hard difficulty levels. """ import random from datetime import datetime, timedelta from typing import Any from env.actions import ActionType from env.state import ( UserContext, LogEntry, NetworkStatus, VPNStatus, AuthStatus, ServiceHealth, TaskLevel ) # ─── Easy Task Root Causes ─────────────────────────────────── TASK_EASY_ROOT_CAUSES = [ { "id": "wifi_adapter_disabled", "description": "Network adapter disabled by OS update", "category": "hardware", "correct_remediation": ActionType.RECONFIGURE_CLIENT, "correct_remediation_alts": [], "diagnostic_path": [ActionType.QUERY_DEVICE_STATUS, ActionType.INSPECT_NETWORK], "severity": "low", "affected_users": 1 }, { "id": "wifi_wrong_credentials", "description": "Incorrect saved credentials for SSID", "category": "auth", "correct_remediation": ActionType.RESET_CREDENTIALS, "correct_remediation_alts": [], "diagnostic_path": [ActionType.CHECK_AUTHENTICATION, ActionType.INSPECT_LOGS], "severity": "low", "affected_users": 1 }, { "id": "wifi_dns_corruption", "description": "Local DNS cache corruption", "category": "dns", "correct_remediation": ActionType.FLUSH_DNS, "correct_remediation_alts": [ActionType.RECONFIGURE_CLIENT], "diagnostic_path": [ActionType.INSPECT_NETWORK, ActionType.INSPECT_LOGS], "severity": "low", "affected_users": 1 }, { "id": "wifi_airplane_mode", "description": "Airplane mode enabled by accident", "category": "hardware", "correct_remediation": ActionType.RECONFIGURE_CLIENT, "correct_remediation_alts": [], "diagnostic_path": [ActionType.QUERY_DEVICE_STATUS], "severity": "low", "affected_users": 1 }, { "id": "wifi_dhcp_exhausted", "description": "DHCP lease pool exhausted", "category": "network", "correct_remediation": ActionType.RESTART_SERVICE, "correct_remediation_alts": [ActionType.ESCALATE_TICKET], "diagnostic_path": [ActionType.INSPECT_NETWORK, ActionType.RUN_DIAGNOSTIC], "severity": "medium", "affected_users": 15 }, { "id": "wifi_wrong_ssid", "description": "Client connected to wrong SSID (guest vs corporate)", "category": "network", "correct_remediation": ActionType.RECONFIGURE_CLIENT, "correct_remediation_alts": [], "diagnostic_path": [ActionType.QUERY_DEVICE_STATUS, ActionType.INSPECT_NETWORK], "severity": "low", "affected_users": 1 }, ] # ─── Medium Task Root Causes ────────────────────────────────── TASK_MEDIUM_ROOT_CAUSES = [ { "id": "expired_password", "description": "User password expired, blocking VPN authentication", "category": "auth", "correct_remediation": ActionType.RESET_CREDENTIALS, "correct_remediation_alts": [ActionType.ESCALATE_TICKET], "diagnostic_path": [ActionType.CHECK_AUTHENTICATION, ActionType.INSPECT_LOGS], "severity": "medium", "affected_users": 1 }, { "id": "vpn_version_mismatch", "description": "VPN client version incompatible with gateway", "category": "vpn", "correct_remediation": ActionType.RECONFIGURE_CLIENT, "correct_remediation_alts": [ActionType.ESCALATE_TICKET], "diagnostic_path": [ActionType.QUERY_DEVICE_STATUS, ActionType.SEARCH_INTERNAL_KB], "severity": "medium", "affected_users": 1 }, { "id": "dns_split_tunnel_conflict", "description": "Split-tunnel DNS configuration conflict", "category": "dns", "correct_remediation": ActionType.FLUSH_DNS, "correct_remediation_alts": [ActionType.RECONFIGURE_CLIENT], "diagnostic_path": [ActionType.INSPECT_NETWORK, ActionType.RUN_DIAGNOSTIC], "severity": "medium", "affected_users": 1 }, { "id": "mfa_token_desync", "description": "MFA token out of sync with authentication server", "category": "auth", "correct_remediation": ActionType.RESET_CREDENTIALS, "correct_remediation_alts": [ActionType.ESCALATE_TICKET], "diagnostic_path": [ActionType.CHECK_AUTHENTICATION, ActionType.CONTACT_USER], "severity": "medium", "affected_users": 1 }, { "id": "firewall_subnet_block", "description": "Firewall blocking specific subnet for remote users", "category": "network", "correct_remediation": ActionType.ESCALATE_TICKET, "correct_remediation_alts": [], "diagnostic_path": [ActionType.INSPECT_NETWORK, ActionType.RUN_DIAGNOSTIC], "severity": "high", "affected_users": 8 }, { "id": "certificate_expiry", "description": "VPN SSL certificate expired on gateway", "category": "vpn", "correct_remediation": ActionType.ESCALATE_TICKET, "correct_remediation_alts": [], "diagnostic_path": [ActionType.INSPECT_LOGS, ActionType.SEARCH_INTERNAL_KB], "severity": "high", "affected_users": 12 }, ] # ─── Hard Task Root Causes ──────────────────────────────────── TASK_HARD_ROOT_CAUSES = [ { "id": "sso_token_service_degradation", "description": "SSO token service degraded in datacenter-east, affecting multiple applications", "category": "sso", "correct_remediation": ActionType.ESCALATE_TICKET, "correct_remediation_alts": [], "diagnostic_path": [ ActionType.INSPECT_LOGS, ActionType.CHECK_AUTHENTICATION, ActionType.RUN_DIAGNOSTIC ], "severity": "critical", "affected_users": 150, "multi_ticket": True }, { "id": "ldap_replication_lag", "description": "LDAP replication lag causing inconsistent auth across regions", "category": "auth", "correct_remediation": ActionType.ESCALATE_TICKET, "correct_remediation_alts": [], "diagnostic_path": [ ActionType.CHECK_AUTHENTICATION, ActionType.INSPECT_LOGS, ActionType.SEARCH_INTERNAL_KB ], "severity": "critical", "affected_users": 200, "multi_ticket": True }, { "id": "dns_zone_transfer_failure", "description": "DNS zone transfer failure causing resolution inconsistencies", "category": "dns", "correct_remediation": ActionType.ESCALATE_TICKET, "correct_remediation_alts": [], "diagnostic_path": [ ActionType.INSPECT_NETWORK, ActionType.INSPECT_LOGS, ActionType.RUN_DIAGNOSTIC ], "severity": "critical", "affected_users": 300, "multi_ticket": True }, { "id": "proxy_cascading_failure", "description": "Web proxy cascading failure affecting multiple services", "category": "network", "correct_remediation": ActionType.ESCALATE_TICKET, "correct_remediation_alts": [], "diagnostic_path": [ ActionType.INSPECT_NETWORK, ActionType.INSPECT_LOGS, ActionType.SEARCH_INTERNAL_KB ], "severity": "critical", "affected_users": 250, "multi_ticket": True }, { "id": "certificate_chain_break", "description": "Intermediate certificate chain break affecting SSL validation", "category": "network", "correct_remediation": ActionType.ESCALATE_TICKET, "correct_remediation_alts": [], "diagnostic_path": [ ActionType.INSPECT_LOGS, ActionType.RUN_DIAGNOSTIC, ActionType.SEARCH_INTERNAL_KB ], "severity": "critical", "affected_users": 180, "multi_ticket": True }, { "id": "storage_backend_degradation", "description": "Shared storage backend degradation affecting file services", "category": "hardware", "correct_remediation": ActionType.ESCALATE_TICKET, "correct_remediation_alts": [], "diagnostic_path": [ ActionType.QUERY_DEVICE_STATUS, ActionType.INSPECT_LOGS, ActionType.RUN_DIAGNOSTIC ], "severity": "critical", "affected_users": 220, "multi_ticket": True }, ] class TicketGenerator: """ Generates realistic IT support tickets with configurable difficulty. Supports adversarial sampling weights for adaptive difficulty. """ def __init__(self, seed: int | None = None): self.rng = random.Random(seed) self._sampling_weights: dict[str, float] = {} def set_sampling_weights(self, weights: dict[str, float]) -> None: """Update sampling weights from adversary.""" self._sampling_weights = weights def generate_ticket( self, task_level: TaskLevel, seed: int | None = None ) -> dict[str, Any]: """ Generate a ticket scenario for the given difficulty level. Returns: Dictionary containing all ticket components: - root_cause: The ground truth root cause info - observation_template: Initial observation data - misleading_log_index: Index of misleading log (if any) """ if seed is not None: self.rng = random.Random(seed) if task_level == TaskLevel.EASY: return self._generate_easy_ticket() elif task_level == TaskLevel.MEDIUM: return self._generate_medium_ticket() elif task_level == TaskLevel.HARD: return self._generate_hard_ticket() else: raise ValueError(f"Invalid task level: {task_level}") def _generate_easy_ticket(self) -> dict[str, Any]: """Generate easy difficulty ticket (Wi-Fi connectivity).""" # Apply sampling weights if available causes = TASK_EASY_ROOT_CAUSES.copy() if self._sampling_weights: weights = [self._sampling_weights.get(c["id"], 1.0) for c in causes] root_cause = self.rng.choices(causes, weights=weights, k=1)[0] else: root_cause = self.rng.choice(causes) # Generate user context user_context = self._generate_user_context() # Generate initial logs (no misleading entries for easy) logs = self._generate_initial_logs_easy(root_cause) # Create observation template observation_template = { "ticket_summary": "Hi, I can't connect to Wi-Fi and need help urgently.", "user_context": user_context, "network_status": NetworkStatus.UNKNOWN, "vpn_status": VPNStatus.NA, "auth_status": AuthStatus.OK, "service_health": { "dhcp": ServiceHealth.UNKNOWN, "dns": ServiceHealth.UNKNOWN, "wifi": ServiceHealth.UNKNOWN, }, "system_logs": logs, } return { "root_cause": root_cause, "observation_template": observation_template, "misleading_log_index": None, } def _generate_medium_ticket(self) -> dict[str, Any]: """Generate medium difficulty ticket (VPN access).""" causes = TASK_MEDIUM_ROOT_CAUSES.copy() if self._sampling_weights: weights = [self._sampling_weights.get(c["id"], 1.0) for c in causes] root_cause = self.rng.choices(causes, weights=weights, k=1)[0] else: root_cause = self.rng.choice(causes) user_context = self._generate_user_context() user_context.location = "remote" # Generate logs with 1 misleading entry logs, misleading_idx = self._generate_initial_logs_medium(root_cause) observation_template = { "ticket_summary": "I'm working from home and can't access any internal tools through VPN. It was working fine yesterday.", "user_context": user_context, "network_status": NetworkStatus.UNKNOWN, "vpn_status": VPNStatus.NA, "auth_status": AuthStatus.OK, "service_health": { "vpn_client": ServiceHealth.UNKNOWN, "vpn_gateway": ServiceHealth.UNKNOWN, "dns": ServiceHealth.UNKNOWN, }, "system_logs": logs, } return { "root_cause": root_cause, "observation_template": observation_template, "misleading_log_index": misleading_idx, } def _generate_hard_ticket(self) -> dict[str, Any]: """Generate hard difficulty ticket (multi-service failure).""" causes = TASK_HARD_ROOT_CAUSES.copy() if self._sampling_weights: weights = [self._sampling_weights.get(c["id"], 1.0) for c in causes] root_cause = self.rng.choices(causes, weights=weights, k=1)[0] else: root_cause = self.rng.choice(causes) user_context = self._generate_user_context() # Generate logs with multiple misleading entries logs, misleading_idx = self._generate_initial_logs_hard(root_cause) observation_template = { "ticket_summary": "Multiple users reporting access issues across different systems. Email and collaboration tools are affected.", "user_context": user_context, "network_status": NetworkStatus.UNKNOWN, "vpn_status": VPNStatus.NA, "auth_status": AuthStatus.OK, "service_health": { "mail": ServiceHealth.UNKNOWN, "sharepoint": ServiceHealth.UNKNOWN, "teams": ServiceHealth.UNKNOWN, "sso": ServiceHealth.UNKNOWN, }, "system_logs": logs, } return { "root_cause": root_cause, "observation_template": observation_template, "misleading_log_index": misleading_idx, } def _generate_user_context(self) -> UserContext: """Generate randomized user context.""" departments = ["Engineering", "Sales", "Finance", "HR", "Operations"] roles = ["Manager", "Analyst", "Developer", "Coordinator", "Specialist"] devices = ["Laptop", "Desktop", "Tablet"] os_versions = ["Windows 11", "Windows 10", "macOS 14", "macOS 13"] locations = ["onsite", "remote"] return UserContext( department=self.rng.choice(departments), role=self.rng.choice(roles), device_type=self.rng.choice(devices), os_version=self.rng.choice(os_versions), location=self.rng.choice(locations), ) def _generate_initial_logs_easy(self, root_cause: dict) -> list[LogEntry]: """Generate initial logs for easy task (no misleading entries).""" now = datetime.utcnow() logs = [] # Add generic startup logs logs.append(LogEntry( timestamp=(now - timedelta(minutes=5)).isoformat(), level="INFO", service="NetworkManager", message="Network service initialized", is_misleading=False )) # Add category-relevant log category = root_cause["category"] if category == "dns": logs.append(LogEntry( timestamp=(now - timedelta(minutes=3)).isoformat(), level="WARN", service="DNSResolver", message="DNS resolution timeout for internal.company.com", is_misleading=False )) elif category == "auth": logs.append(LogEntry( timestamp=(now - timedelta(minutes=2)).isoformat(), level="ERROR", service="WiFiAuth", message="Authentication failed: Invalid credentials", is_misleading=False )) elif category == "hardware": logs.append(LogEntry( timestamp=(now - timedelta(minutes=1)).isoformat(), level="WARN", service="DeviceManager", message="Network adapter state changed", is_misleading=False )) logs.append(LogEntry( timestamp=now.isoformat(), level="ERROR", service="NetworkManager", message="Connection attempt failed", is_misleading=False )) return logs def _generate_initial_logs_medium( self, root_cause: dict ) -> tuple[list[LogEntry], int]: """Generate logs for medium task with 1 misleading entry.""" now = datetime.utcnow() logs = [] # Add real logs logs.append(LogEntry( timestamp=(now - timedelta(minutes=10)).isoformat(), level="INFO", service="VPNClient", message="VPN client started", is_misleading=False )) category = root_cause["category"] if category == "auth": logs.append(LogEntry( timestamp=(now - timedelta(minutes=7)).isoformat(), level="ERROR", service="AuthService", message=f"Authentication failed: {root_cause['description']}", is_misleading=False )) elif category == "vpn": logs.append(LogEntry( timestamp=(now - timedelta(minutes=6)).isoformat(), level="WARN", service="VPNGateway", message=f"VPN handshake failed: {root_cause['description']}", is_misleading=False )) # Add misleading log (at random position) misleading_messages = [ "Network latency spike detected (250ms avg)", "DNS query timeout for vpn.company.com", "Firewall rule evaluation took longer than expected", ] misleading_log = LogEntry( timestamp=(now - timedelta(minutes=4)).isoformat(), level="WARN", service="NetworkMonitor", message=self.rng.choice(misleading_messages), is_misleading=True ) # Insert misleading log at random position (not first or last) insert_pos = self.rng.randint(1, len(logs)) logs.insert(insert_pos, misleading_log) logs.append(LogEntry( timestamp=now.isoformat(), level="ERROR", service="VPNClient", message="Connection failed after 3 retries", is_misleading=False )) return logs, insert_pos def _generate_initial_logs_hard( self, root_cause: dict ) -> tuple[list[LogEntry], int | None]: """Generate logs for hard task with multiple misleading entries.""" now = datetime.utcnow() logs = [] # Add system logs with contradictory evidence logs.append(LogEntry( timestamp=(now - timedelta(minutes=15)).isoformat(), level="INFO", service="SystemMonitor", message="All services nominal", is_misleading=True # Contradicts the actual failures )) logs.append(LogEntry( timestamp=(now - timedelta(minutes=12)).isoformat(), level="WARN", service="LoadBalancer", message="Backend pool health check degraded", is_misleading=False )) # Add a misleading authentication success logs.append(LogEntry( timestamp=(now - timedelta(minutes=10)).isoformat(), level="INFO", service="AuthService", message="User authentication successful (cached token)", is_misleading=True )) category = root_cause["category"] if category == "sso": logs.append(LogEntry( timestamp=(now - timedelta(minutes=8)).isoformat(), level="ERROR", service="SSOTokenService", message="Token validation latency exceeding threshold (3000ms)", is_misleading=False )) elif category == "dns": logs.append(LogEntry( timestamp=(now - timedelta(minutes=7)).isoformat(), level="ERROR", service="DNSService", message="Zone transfer failed from primary nameserver", is_misleading=False )) logs.append(LogEntry( timestamp=(now - timedelta(minutes=5)).isoformat(), level="WARN", service="ServiceMesh", message="Multiple service endpoints reporting degraded state", is_misleading=False )) logs.append(LogEntry( timestamp=now.isoformat(), level="ERROR", service="IncidentManager", message=f"Multiple concurrent failures detected: {root_cause['description']}", is_misleading=False )) # Return index of first misleading log misleading_idx = 0 # First log is misleading return logs, misleading_idx