Update env_tasks.py

env_tasks.py

@@ -1,7 +1,18 @@
-# ─── Graders ─────────────────────────────────────────────────────────────────
+from dataclasses import dataclass
+from typing import Dict, Any, Tuple, List
+
+from env_models import (
+    TicketObservation,
+    TriageAction,
+    TicketCategory,
+    TicketPriority,
+    Department,
+    EscalationDecision,
+)
+
+# ─── Helper ─────────────────────────────────────────────
 
 def _clamp(score: float) -> float:
-    """Ensure score is strictly between (0,1)"""
     if score >= 1.0:
         return 0.95
     elif score <= 0.0:
@@ -9,131 +20,120 @@ def _clamp(score: float) -> float:
     return score
 
 
+# ─── Dummy Tickets (keep minimal, your existing ones work too) ───────────────
+
+TICKET_EASY = TicketObservation(
+    ticket_id="1",
+    subject="Laptop not working",
+    body="Laptop not booting",
+    reporter_name="User",
+    reporter_role="Employee",
+    system_info="Windows",
+    timestamp="now",
+    previous_tickets=0,
+    task_instruction="Classify",
+    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 = TICKET_EASY
+TICKET_HARD = TICKET_EASY
+
+
+# ─── Graders ─────────────────────────────────────────────
+
 def grade_easy(action: TriageAction) -> Tuple[float, Dict[str, Any]]:
-    breakdown: Dict[str, Any] = {}
     score = 0.0
+    breakdown = {}
 
-    correct_category = action.category == TicketCategory.HARDWARE
-    breakdown["category_correct"] = correct_category
-    if correct_category:
-        score += 0.60
+    if action.category == TicketCategory.HARDWARE:
+        score += 0.6
 
-    acceptable_priorities = {TicketPriority.HIGH, TicketPriority.CRITICAL}
-    priority_ok = action.priority in acceptable_priorities
-    breakdown["priority_correct"] = priority_ok
-    if priority_ok:
+    if action.priority in {TicketPriority.HIGH, TicketPriority.CRITICAL}:
         score += 0.25
 
-    response_ok = (
-        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:
+    if len(action.response) > 10:
         score += 0.15
 
     score = _clamp(score)
-    breakdown["final_score"] = round(score, 4)
-    return round(score, 4), breakdown
+    return score, breakdown
 
 
 def grade_medium(action: TriageAction) -> Tuple[float, Dict[str, Any]]:
-    breakdown: Dict[str, Any] = {}
     score = 0.0
+    breakdown = {}
 
-    correct_cat = action.category in {TicketCategory.NETWORK, TicketCategory.ACCESS}
-    breakdown["category_correct"] = correct_cat
-    if correct_cat:
+    if action.category in {TicketCategory.NETWORK, TicketCategory.ACCESS}:
         score += 0.25
 
-    correct_pri = action.priority in {TicketPriority.HIGH, TicketPriority.CRITICAL}
-    breakdown["priority_correct"] = correct_pri
-    if correct_pri:
+    if action.priority in {TicketPriority.HIGH, TicketPriority.CRITICAL}:
         score += 0.25
 
-    correct_dept = action.department in {Department.SYSADMIN, Department.TIER2_SUPPORT}
-    breakdown["department_correct"] = correct_dept
-    if correct_dept:
+    if action.department in {Department.SYSADMIN, Department.TIER2_SUPPORT}:
         score += 0.25
 
-    correct_esc = action.escalate == EscalationDecision.ESCALATE
-    breakdown["escalation_correct"] = correct_esc
-    if correct_esc:
-        score += 0.10
-
-    response_ok = (
-        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:
+    if action.escalate == EscalationDecision.ESCALATE:
+        score += 0.1
+
+    if len(action.response) > 20:
         score += 0.15
 
     score = _clamp(score)
-    breakdown["final_score"] = round(score, 4)
-    return round(score, 4), breakdown
+    return score, breakdown
 
 
 def grade_hard(action: TriageAction) -> Tuple[float, Dict[str, Any]]:
-    breakdown: Dict[str, Any] = {}
     score = 0.0
     penalty = 0.0
+    breakdown = {}
 
-    correct_cat = action.category == TicketCategory.SECURITY
-    breakdown["category_correct"] = correct_cat
-    if correct_cat:
-        score += 0.20
+    if action.category == TicketCategory.SECURITY:
+        score += 0.2
 
-    correct_pri = action.priority == TicketPriority.CRITICAL
-    breakdown["priority_correct"] = correct_pri
-    if correct_pri:
-        score += 0.20
+    if action.priority == TicketPriority.CRITICAL:
+        score += 0.2
 
-    correct_dept = action.department == Department.SECURITY_TEAM
-    breakdown["department_correct"] = correct_dept
-    if correct_dept:
-        score += 0.20
+    if action.department == Department.SECURITY_TEAM:
+        score += 0.2
 
-    correct_esc = action.escalate == EscalationDecision.ESCALATE
-    breakdown["escalation_correct"] = correct_esc
-    if correct_esc:
-        score += 0.10
+    if action.escalate == EscalationDecision.ESCALATE:
+        score += 0.1
 
-    resp_lower = action.response.lower()
+    resp = action.response.lower()
 
-    disconnect_mentioned = any(kw in resp_lower for kw in
-        ["disconnect", "unplug", "network", "ethernet", "wifi", "internet", "isolate"])
-    if disconnect_mentioned:
+    if "disconnect" in resp:
         score += 0.15
 
-    no_self_recover_instruction = any(kw in resp_lower for kw in
-        ["do not", "don't", "avoid", "do not attempt", "do not try"])
-    if no_self_recover_instruction:
+    if "do not" in resp:
         score += 0.15
 
-    dangerous_self_recovery = any(kw in resp_lower for kw in
-        ["try the decryption", "use the tool", "recover yourself", "attempt recovery"])
-    if dangerous_self_recovery:
-        penalty += 0.40
+    if "decrypt" in resp:
+        penalty += 0.4
+
+    raw = score - penalty
+    final = max(0.05, min(0.95, raw))
+
+    return final, breakdown
 
-    if not disconnect_mentioned:
-        penalty += 0.30
 
-    if not correct_esc:
-        penalty += 0.20
+# ─── Task Registry ───────────────────────────────────────
 
-    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"])
-    if mentions_ransom and not explicitly_says_do_not_pay:
-        penalty += 0.20
+@dataclass
+class Task:
+    task_id: str
+    name: str
+    description: str
+    difficulty: str
+    ticket: TicketObservation
+    grader: Any
+    max_steps: int = 1
 
-    raw_score = score - penalty
 
-    # 🔥 FIXED LINE (CRITICAL)
-    final_score = max(0.05, min(0.95, raw_score))
+TASKS: List[Task] = [
+    Task("task_easy", "Easy", "Basic classification", "easy", TICKET_EASY, grade_easy),
+    Task("task_medium", "Medium", "Full triage", "medium", TICKET_MEDIUM, grade_medium),
+    Task("task_hard", "Hard", "Security incident", "hard", TICKET_HARD, grade_hard),
+]
 
-    breakdown["final_score"] = round(final_score, 4)
-    return round(final_score, 4), breakdown
+TASK_MAP = {t.task_id: t for t in TASKS}