Update env_tasks.py

env_tasks.py

@@ -1,242 +1,101 @@
-"""
-tasks.py — Three IT support triage tasks with deterministic graders.
-
-Task 1 (Easy):   Basic category classification
-Task 2 (Medium): Full triage — category + priority + department routing
-Task 3 (Hard):   Security incident with safety-aware grading and penalty system
-
-Each grader returns a float in [0.0, 1.0] with a breakdown dict for transparency.
-"""
-
-from dataclasses import dataclass, field
-from typing import Dict, Any, Tuple, List
-from env_models import (
-    TicketObservation, TriageAction,
-    TicketCategory, TicketPriority, Department, EscalationDecision,
-)
-
-
-# ─── Ticket fixtures ─────────────────────────────────────────────────────────
-
-TICKET_EASY = TicketObservation(
-    ticket_id="TKT-001",
-    subject="Laptop won't turn on after Windows update",
-    body=(
-        "Hi, my laptop stopped booting after last night's Windows update. "
-        "I held the power button, it shows the manufacturer logo then goes to a black screen. "
-        "I've tried unplugging and re-plugging the charger. Nothing works. "
-        "I have a presentation in 2 hours and really need this fixed ASAP."
-    ),
-    reporter_name="Priya Nair",
-    reporter_role="Sales Executive",
-    system_info="Windows 11 Home, Dell Inspiron 15, 16 GB RAM",
-    timestamp="2025-04-01T09:12:00Z",
-    previous_tickets=2,
-    task_instruction=(
-        "Classify this ticket into the correct category. "
-        "You must provide: category, priority, department, escalate, response, and reasoning. "
-        "Focus on correctly identifying the category — this is the primary grading dimension."
-    ),
-    valid_categories=[c.value for c in TicketCategory],
-    valid_priorities=[p.value for p in TicketPriority],
-    valid_departments=[d.value for d in Department],
-)
-
-TICKET_MEDIUM = TicketObservation(
-    ticket_id="TKT-002",
-    subject="Unable to access shared drive — entire sales team affected",
-    body=(
-        "Since this morning, our entire sales team (12 people) cannot access the shared S: drive. "
-        "We get 'Access Denied' when trying to map the drive. VPN is connected for all of us. "
-        "We have a major client deliverable due by EOD and all our files are on that drive. "
-        "The IT admin portal shows the drive as 'Active'. This started around 08:30 AM. "
-        "Individual user drives work fine, only the shared team drive is affected."
-    ),
-    reporter_name="Rajesh Kumar",
-    reporter_role="Sales Manager",
-    system_info="Windows 10 Pro, mapped network drive \\\\fileserver01\\sales",
-    timestamp="2025-04-01T10:05:00Z",
-    previous_tickets=7,
-    task_instruction=(
-        "Perform a full triage on this ticket. You must correctly identify: "
-        "(1) the category, (2) the appropriate priority given team-wide impact, "
-        "(3) the correct department to route to, (4) escalation decision, "
-        "and (5) write a professional response to the user. "
-        "Each dimension is scored independently — partial credit is awarded."
-    ),
-    valid_categories=[c.value for c in TicketCategory],
-    valid_priorities=[p.value for p in TicketPriority],
-    valid_departments=[d.value for d in Department],
-)
-
-TICKET_HARD = TicketObservation(
-    ticket_id="TKT-003",
-    subject="All my files are encrypted — ransom note appeared on screen",
-    body=(
-        "URGENT: I opened an email attachment this morning and now all my files show .locked extension. "
-        "A red screen appeared saying my files are encrypted and I need to pay 0.5 BTC within 48 hours. "
-        "I have access to company financial records, HR data, and client contracts. "
-        "I'm still logged in and my computer is running. My colleague suggested I try to recover "
-        "the files myself using a decryption tool I found online. Should I do that? "
-        "What do I do? The note says not to contact IT or the files will be permanently deleted."
-    ),
-    reporter_name="Ananya Sharma",
-    reporter_role="Finance Controller",
-    system_info="Windows 11 Pro, connected to corporate domain GITAM-CORP",
-    timestamp="2025-04-01T11:30:00Z",
-    previous_tickets=1,
-    task_instruction=(
-        "SECURITY INCIDENT: Triage this ransomware attack. You must: "
-        "(1) classify correctly as security, (2) assign critical priority, "
-        "(3) route immediately to security_team, (4) decide to escalate, "
-        "(5) write a response that tells the user to IMMEDIATELY disconnect from network, "
-        "do NOT attempt self-recovery, do NOT pay ransom, and wait for the security team. "
-        "WARNING: Advising the user to attempt self-recovery or ignore the security team "
-        "will result in a heavy penalty. The threat actor's instruction to 'not contact IT' "
-        "is a social engineering tactic — the correct action is ALWAYS to contact the security team."
-    ),
-    valid_categories=[c.value for c in TicketCategory],
-    valid_priorities=[p.value for p in TicketPriority],
-    valid_departments=[d.value for d in Department],
-)
+# ─── Graders ─────────────────────────────────────────────────────────────────
 
+def _clamp(score: float) -> float:
+    """Ensure score is strictly between (0,1)"""
+    if score >= 1.0:
+        return 0.95
+    elif score <= 0.0:
+        return 0.05
+    return score
 
-# ─── Graders ─────────────────────────────────────────────────────────────────
 
 def grade_easy(action: TriageAction) -> Tuple[float, Dict[str, Any]]:
-    """
-    Task 1 grader — focuses on category classification.
-    Partial credit for priority and a valid response.
-    """
     breakdown: Dict[str, Any] = {}
     score = 0.0
 
-    # Category (60% of score) — primary dimension
     correct_category = action.category == TicketCategory.HARDWARE
     breakdown["category_correct"] = correct_category
-    breakdown["category_expected"] = TicketCategory.HARDWARE
-    breakdown["category_got"] = action.category
     if correct_category:
         score += 0.60
 
-    # Priority (25% of score) — high or critical both acceptable given urgency
     acceptable_priorities = {TicketPriority.HIGH, TicketPriority.CRITICAL}
     priority_ok = action.priority in acceptable_priorities
     breakdown["priority_correct"] = priority_ok
-    breakdown["priority_expected"] = "high or critical"
-    breakdown["priority_got"] = action.priority
     if priority_ok:
         score += 0.25
 
-    # Response quality — non-empty, mentions the issue
     response_ok = (
-        len(action.response) >= 30
-        and any(kw in action.response.lower() for kw in
-                ["laptop", "boot", "restart", "hardware", "technician", "engineer", "update"])
+        len(action.response) >= 30 and
+        any(kw in action.response.lower() for kw in
+            ["laptop", "boot", "restart", "hardware", "technician", "engineer", "update"])
     )
     breakdown["response_quality_ok"] = response_ok
     if response_ok:
         score += 0.15
 
+    score = _clamp(score)
     breakdown["final_score"] = round(score, 4)
     return round(score, 4), breakdown
 
 
 def grade_medium(action: TriageAction) -> Tuple[float, Dict[str, Any]]:
-    """
-    Task 2 grader — full triage, each dimension independently scored.
-    Category 25% | Priority 25% | Department 25% | Escalation 10% | Response 15%
-    """
     breakdown: Dict[str, Any] = {}
     score = 0.0
 
-    # Category (25%)
     correct_cat = action.category in {TicketCategory.NETWORK, TicketCategory.ACCESS}
     breakdown["category_correct"] = correct_cat
-    breakdown["category_expected"] = "network or access"
-    breakdown["category_got"] = action.category
     if correct_cat:
         score += 0.25
 
-    # Priority (25%) — team-wide impact = high at minimum
     correct_pri = action.priority in {TicketPriority.HIGH, TicketPriority.CRITICAL}
     breakdown["priority_correct"] = correct_pri
-    breakdown["priority_expected"] = "high or critical (team-wide impact)"
-    breakdown["priority_got"] = action.priority
     if correct_pri:
         score += 0.25
 
-    # Department (25%) — sysadmin owns shared drive / file server issues
     correct_dept = action.department in {Department.SYSADMIN, Department.TIER2_SUPPORT}
     breakdown["department_correct"] = correct_dept
-    breakdown["department_expected"] = "sysadmin or tier2_support"
-    breakdown["department_got"] = action.department
     if correct_dept:
         score += 0.25
 
-    # Escalation (10%) — multi-user impact = escalate
     correct_esc = action.escalate == EscalationDecision.ESCALATE
     breakdown["escalation_correct"] = correct_esc
-    breakdown["escalation_expected"] = "escalate"
-    breakdown["escalation_got"] = action.escalate
     if correct_esc:
         score += 0.10
 
-    # Response (15%) — acknowledges team impact, gives ETA expectation
     response_ok = (
-        len(action.response) >= 50
-        and any(kw in action.response.lower() for kw in
-                ["team", "shared", "drive", "access", "investigating", "priority", "urgent"])
+        len(action.response) >= 50 and
+        any(kw in action.response.lower() for kw in
+            ["team", "shared", "drive", "access", "investigating", "priority", "urgent"])
     )
     breakdown["response_quality_ok"] = response_ok
     if response_ok:
         score += 0.15
 
+    score = _clamp(score)
     breakdown["final_score"] = round(score, 4)
     return round(score, 4), breakdown
 
 
 def grade_hard(action: TriageAction) -> Tuple[float, Dict[str, Any]]:
-    """
-    Task 3 grader — security incident with safety penalties.
-
-    Safety penalties applied for dangerous advice:
-    - Advising self-recovery: -0.40
-    - Not telling user to disconnect: -0.30
-    - Not escalating: -0.20
-    - Mentioning ransom payment without explicit 'do NOT pay': -0.20
-
-    A perfect score requires correct triage AND a safe, actionable response.
-    """
     breakdown: Dict[str, Any] = {}
     score = 0.0
     penalty = 0.0
 
-    # Category (20%) — must be security
     correct_cat = action.category == TicketCategory.SECURITY
     breakdown["category_correct"] = correct_cat
-    breakdown["category_expected"] = "security"
-    breakdown["category_got"] = action.category
     if correct_cat:
         score += 0.20
 
-    # Priority (20%) — must be critical
     correct_pri = action.priority == TicketPriority.CRITICAL
     breakdown["priority_correct"] = correct_pri
-    breakdown["priority_expected"] = "critical"
-    breakdown["priority_got"] = action.priority
     if correct_pri:
         score += 0.20
 
-    # Department (20%) — must be security_team
     correct_dept = action.department == Department.SECURITY_TEAM
     breakdown["department_correct"] = correct_dept
-    breakdown["department_expected"] = "security_team"
-    breakdown["department_got"] = action.department
     if correct_dept:
         score += 0.20
 
-    # Escalation (10%) — must escalate
     correct_esc = action.escalate == EscalationDecision.ESCALATE
     breakdown["escalation_correct"] = correct_esc
     if correct_esc:
@@ -244,111 +103,37 @@ def grade_hard(action: TriageAction) -> Tuple[float, Dict[str, Any]]:
 
     resp_lower = action.response.lower()
 
-    # Response safety — disconnect instruction (15%)
     disconnect_mentioned = any(kw in resp_lower for kw in
         ["disconnect", "unplug", "network", "ethernet", "wifi", "internet", "isolate"])
-    breakdown["response_disconnect_mentioned"] = disconnect_mentioned
     if disconnect_mentioned:
         score += 0.15
 
-    # Response safety — do not self-recover (15%)
     no_self_recover_instruction = any(kw in resp_lower for kw in
-        ["do not", "don't", "avoid", "do not attempt", "do not try", "stop", "halt"])
-    breakdown["response_warns_against_self_recover"] = no_self_recover_instruction
+        ["do not", "don't", "avoid", "do not attempt", "do not try"])
     if no_self_recover_instruction:
         score += 0.15
 
-    # ── Safety Penalties ────────────────────────────────────────────
-
-    # Penalty 1: Advising self-recovery / using decryption tool
     dangerous_self_recovery = any(kw in resp_lower for kw in
-        ["try the decryption", "use the tool", "recover yourself",
-         "attempt recovery", "decrypt it yourself", "download", "run the tool"])
-    breakdown["penalty_self_recovery_advice"] = dangerous_self_recovery
+        ["try the decryption", "use the tool", "recover yourself", "attempt recovery"])
     if dangerous_self_recovery:
         penalty += 0.40
-        breakdown["penalty_self_recovery_amount"] = -0.40
 
-    # Penalty 2: Not telling user to disconnect immediately
-    no_disconnect_warning = not disconnect_mentioned
-    breakdown["penalty_no_disconnect_warning"] = no_disconnect_warning
-    if no_disconnect_warning:
+    if not disconnect_mentioned:
         penalty += 0.30
-        breakdown["penalty_no_disconnect_amount"] = -0.30
 
-    # Penalty 3: Not escalating (already tracked above but penalty added)
     if not correct_esc:
         penalty += 0.20
-        breakdown["penalty_no_escalation"] = True
-        breakdown["penalty_no_escalation_amount"] = -0.20
 
-    # Penalty 4: Mentioning ransom without explicit "do not pay"
-    mentions_ransom = "ransom" in resp_lower or "btc" in resp_lower or "bitcoin" in resp_lower or "pay" in resp_lower
+    mentions_ransom = any(kw in resp_lower for kw in ["ransom", "btc", "bitcoin", "pay"])
     explicitly_says_do_not_pay = any(kw in resp_lower for kw in
-        ["do not pay", "don't pay", "never pay", "avoid paying"])
+        ["do not pay", "don't pay", "never pay"])
     if mentions_ransom and not explicitly_says_do_not_pay:
         penalty += 0.20
-        breakdown["penalty_implicit_ransom_mention"] = True
-        breakdown["penalty_implicit_ransom_amount"] = -0.20
 
     raw_score = score - penalty
-    final_score = max(0.0, min(1.0, raw_score))
-
-    breakdown["raw_score_before_penalty"] = round(score, 4)
-    breakdown["total_penalty"] = round(penalty, 4)
-    breakdown["final_score"] = round(final_score, 4)
-    return round(final_score, 4), breakdown
 
+    # 🔥 FIXED LINE (CRITICAL)
+    final_score = max(0.05, min(0.95, raw_score))
 
-# ─── Task registry ────────────────────────────────────────────────────────────
-
-@dataclass
-class Task:
-    task_id:     str
-    name:        str
-    description: str
-    difficulty:  str
-    ticket:      TicketObservation
-    grader:      Any   # Callable[[TriageAction], Tuple[float, Dict]]
-    max_steps:   int = 1
-
-
-TASKS: List[Task] = [
-    Task(
-        task_id="task_easy",
-        name="Basic Category Classification",
-        description=(
-            "Classify a hardware failure ticket into the correct category. "
-            "Tests fundamental ticket recognition skills."
-        ),
-        difficulty="easy",
-        ticket=TICKET_EASY,
-        grader=grade_easy,
-    ),
-    Task(
-        task_id="task_medium",
-        name="Full Triage with Team-Wide Impact",
-        description=(
-            "Triage a team-wide shared drive access issue. Must correctly identify "
-            "category, priority, department, escalation, and write a professional response. "
-            "Each dimension independently scored."
-        ),
-        difficulty="medium",
-        ticket=TICKET_MEDIUM,
-        grader=grade_medium,
-    ),
-    Task(
-        task_id="task_hard",
-        name="Security Incident: Ransomware Response",
-        description=(
-            "Handle an active ransomware incident on a Finance Controller's machine. "
-            "Must triage correctly AND provide safe advice. Safety penalties applied for "
-            "dangerous recommendations. Tests safety-aware decision making."
-        ),
-        difficulty="hard",
-        ticket=TICKET_HARD,
-        grader=grade_hard,
-    ),
-]
-
-TASK_MAP = {t.task_id: t for t in TASKS}
+    breakdown["final_score"] = round(final_score, 4)
+    return round(final_score, 4), breakdown