Spaces:
Sleeping
Sleeping
| """Reusable policy helpers for local baselines and training examples.""" | |
| from __future__ import annotations | |
| from models import SupportDeskAction, SupportDeskObservation | |
| from tasks import get_task | |
| def default_reply(task_id: str) -> str: | |
| """Return a task-specific high-signal customer reply.""" | |
| if task_id == "billing_refund_easy": | |
| return ( | |
| "Thanks for flagging the duplicate charge. I have started the refund for the extra " | |
| "charge, and the funds usually appear within 5-7 business days." | |
| ) | |
| if task_id == "account_takeover_medium": | |
| return ( | |
| "We have escalated this to our trust team. Please reset your password, scan your " | |
| "device for malware, and reply with your workspace_id, last successful login time, " | |
| "and billing email so we can verify the account safely." | |
| ) | |
| if task_id == "regulated_export_exception_hard": | |
| return ( | |
| "We cannot provide a bypass or temporary unlock yet. Our compliance team is running " | |
| "a compliance review, and we need your tenant_region, dpa_amendment_id, and " | |
| "legal_contact_email to continue that review." | |
| ) | |
| return ( | |
| "We are treating this as an active incident and our on-call engineering team is engaged. " | |
| "Please send the affected request IDs, UTC timestamps, and the impacted region so we can " | |
| "speed up the investigation." | |
| ) | |
| def default_note(task_id: str) -> str: | |
| """Return a task-specific internal note.""" | |
| if task_id == "billing_refund_easy": | |
| return "Duplicate charge confirmed from attached invoice; refund approved." | |
| if task_id == "account_takeover_medium": | |
| return "Suspicious login alert reported and customer is locked out." | |
| if task_id == "regulated_export_exception_hard": | |
| return ( | |
| "Audit-driven export exception request tied to an EU residency policy block; " | |
| "customer asked for a manual bypass before legal approval." | |
| ) | |
| return "EU data residency rollout hit intermittent HTTP 500s and the customer launches tonight." | |
| def heuristic_action(observation: SupportDeskObservation) -> SupportDeskAction: | |
| """Deterministic high-performing policy used by the baseline.""" | |
| task = get_task(observation.task_id) | |
| case = observation.case | |
| if case.queue is None or case.priority is None or case.issue_type is None: | |
| return SupportDeskAction( | |
| operation="classify", | |
| queue=task.gold_queue, | |
| priority=task.gold_priority, | |
| issue_type=task.gold_issue_type, | |
| ) | |
| if task.required_requested_fields and sorted(case.requested_fields) != sorted(task.required_requested_fields): | |
| return SupportDeskAction( | |
| operation="request_info", | |
| requested_fields=list(task.required_requested_fields), | |
| ) | |
| if case.customer_follow_up.status == "pending": | |
| return SupportDeskAction(operation="wait") | |
| if not case.reply: | |
| return SupportDeskAction(operation="draft_reply", reply=default_reply(observation.task_id)) | |
| if not case.internal_note: | |
| return SupportDeskAction(operation="add_internal_note", internal_note=default_note(observation.task_id)) | |
| return SupportDeskAction( | |
| operation="submit", | |
| status=task.gold_status, | |
| resolution_code=task.gold_resolution_code, | |
| ) | |