from __future__ import annotations from collections import OrderedDict from dataclasses import dataclass, field from typing import Literal BENCHMARK_NAME = "customer_support_ticket_handler" DEFAULT_SUCCESS_THRESHOLD = 0.75 RESET_URL = "https://app.acmecloud.com/reset" @dataclass(frozen=True) class KnowledgeBaseArticle: article_id: str title: str tags: tuple[str, ...] snippet: str content: str @dataclass(frozen=True) class TicketFixture: ticket_id: str customer_id: str organization_name: str subject: str message: str @dataclass(frozen=True) class AccountFixture: customer_id: str organization_name: str plan: str tenure_years: float | None = None arr_usd: int | None = None duplicate_charge_amount_cents: int | None = None duplicate_charge_count: int = 0 duplicate_charge_refund_eligible: bool = False legal_threat: bool = False incident_severity: str | None = None mandatory_escalation_queue: str | None = None mandatory_escalation_priority: str | None = None @dataclass(frozen=True) class TaskFixture: task_id: str title: str difficulty: Literal["easy", "medium", "hard"] ticket: TicketFixture account: AccountFixture relevant_kb_article_id: str | None = None expected_terminal_mode: Literal["resolve", "escalate"] = "resolve" expected_resolution_code: str | None = None expected_refund_amount_cents: int | None = None refund_reason_code: str | None = None expected_escalation_queue: str | None = None expected_escalation_priority: str | None = None reply_keyword_groups: dict[str, tuple[str, ...]] = field(default_factory=dict) forbidden_reply_phrases: tuple[str, ...] = () rubric_weights: dict[str, float] = field(default_factory=dict) efficiency_bonus_max_steps: int | None = None KB_ARTICLES = OrderedDict( ( article.article_id, article, ) for article in ( KnowledgeBaseArticle( article_id="KB-PW-RESET", title="Password reset email troubleshooting", tags=("password", "reset", "email", "spam", "login"), snippet="Ask the customer to use the AcmeCloud reset page, check spam or junk, and wait 5 minutes before retrying.", content=( f"If a password reset email does not arrive, direct the user to {RESET_URL}. " "Ask them to check their spam or junk folder and wait 5 minutes before requesting another email." ), ), KnowledgeBaseArticle( article_id="KB-BILL-DUPLICATE", title="Duplicate subscription charge refund policy", tags=("billing", "refund", "duplicate", "charge", "subscription"), snippet="After verifying a duplicate charge, support can refund the extra charge. Refunds settle in 3-5 business days.", content=( "If account history confirms an accidental duplicate subscription charge, refund the duplicate amount in full. " "Communicate that the refund will appear in 3-5 business days." ), ), KnowledgeBaseArticle( article_id="KB-INCIDENT-LEGAL", title="Critical data incident legal escalation", tags=("incident", "legal", "data", "escalation", "enterprise"), snippet="Legal threats and alleged customer data loss must be escalated immediately to the legal_data_incident queue at P0.", content=( "If an enterprise customer reports data loss and mentions legal action, do not promise a resolution, do not admit fault, " "and escalate immediately to the legal_data_incident queue with priority P0." ), ), KnowledgeBaseArticle( article_id="KB-SSO-SETUP", title="Single sign-on setup guide", tags=("sso", "setup", "identity", "onboarding"), snippet="Configure SAML or OIDC before enforcing SSO in production.", content="SSO setup steps for administrators integrating AcmeCloud with their identity provider.", ), KnowledgeBaseArticle( article_id="KB-INVOICE-DOWNLOAD", title="Invoice download instructions", tags=("invoice", "billing", "download", "finance"), snippet="Billing administrators can download invoices from the Finance tab in workspace settings.", content="Steps for locating billing history and downloading invoices from the AcmeCloud admin console.", ), KnowledgeBaseArticle( article_id="KB-MFA-RESET", title="Multi-factor authentication reset", tags=("mfa", "reset", "login", "security"), snippet="MFA resets require identity verification or admin override.", content="How to reset multi-factor authentication for locked-out users.", ), ) ) TASK_FIXTURES = OrderedDict( ( task.task_id, task, ) for task in ( TaskFixture( task_id="password_reset_guidance", title="Password reset guidance", difficulty="easy", ticket=TicketFixture( ticket_id="ticket_pw_001", customer_id="cust_pw_001", organization_name="Northstar Analytics", subject="Reset email never arrived", message="Hi, I forgot my password and the reset email isn't arriving. Please help.", ), account=AccountFixture( customer_id="cust_pw_001", organization_name="Northstar Analytics", plan="Pro", ), relevant_kb_article_id="KB-PW-RESET", expected_terminal_mode="resolve", expected_resolution_code="password_reset_guidance", reply_keyword_groups={ "reset_url": (RESET_URL,), "spam_folder": ("spam", "junk"), "wait_guidance": ("5 minutes", "five minutes"), }, rubric_weights={ "searched_kb": 0.20, "reply_has_reset_url": 0.30, "reply_mentions_spam_folder": 0.20, "resolved_correctly": 0.20, "efficient_completion": 0.10, }, efficiency_bonus_max_steps=4, ), TaskFixture( task_id="duplicate_charge_refund", title="Duplicate subscription charge refund", difficulty="medium", ticket=TicketFixture( ticket_id="ticket_bill_002", customer_id="cust_bill_002", organization_name="BlueOrbit Labs", subject="Charged twice this month", message=( "I was charged twice for my subscription this month. I want both charges refunded immediately. " "I've been a customer for 3 years." ), ), account=AccountFixture( customer_id="cust_bill_002", organization_name="BlueOrbit Labs", plan="Business", tenure_years=3.2, duplicate_charge_amount_cents=4900, duplicate_charge_count=2, duplicate_charge_refund_eligible=True, ), relevant_kb_article_id="KB-BILL-DUPLICATE", expected_terminal_mode="resolve", expected_resolution_code="billing_refund_processed", expected_refund_amount_cents=4900, refund_reason_code="duplicate_charge", reply_keyword_groups={ "timeline": ("3-5 business days", "3 to 5 business days"), "duplicate_ack": ("duplicate charge", "double charge", "charged twice"), "regret": ("sorry", "apologize", "regret"), "refund_confirmed": ("refund", "refunded", "processed"), }, rubric_weights={ "lookup_account": 0.15, "searched_kb": 0.15, "correct_refund": 0.25, "reply_mentions_timeline": 0.20, "reply_acknowledges_and_apologizes": 0.15, "resolved_without_escalation": 0.10, }, ), TaskFixture( task_id="enterprise_data_loss_escalation", title="Enterprise data loss legal escalation", difficulty="hard", ticket=TicketFixture( ticket_id="ticket_ent_003", customer_id="cust_ent_003", organization_name="Granite Peak Holdings", subject="Critical enterprise incident", message=( "Your platform deleted 2 years of our customer data during last night's maintenance. " "We are a Fortune 500 client. Our legal team will be in contact unless this is resolved in 2 hours. " "I need to speak to your CTO immediately." ), ), account=AccountFixture( customer_id="cust_ent_003", organization_name="Granite Peak Holdings", plan="Enterprise", arr_usd=500000, legal_threat=True, incident_severity="data_loss", mandatory_escalation_queue="legal_data_incident", mandatory_escalation_priority="P0", ), relevant_kb_article_id="KB-INCIDENT-LEGAL", expected_terminal_mode="escalate", expected_escalation_queue="legal_data_incident", expected_escalation_priority="P0", reply_keyword_groups={ "urgency": ("urgent", "immediately", "right away", "priority"), "escalation": ("escalating", "escalated", "investigated", "investigation"), }, forbidden_reply_phrases=( "we deleted your data", "this is our fault", "we are liable", "we caused this", "we guarantee recovery", ), rubric_weights={ "lookup_account": 0.10, "no_refund_or_policy_action": 0.20, "reply_sent_before_escalation": 0.15, "careful_reply": 0.20, "correct_escalation": 0.25, "not_resolved": 0.10, }, ), ) ) def list_task_ids() -> list[str]: return list(TASK_FIXTURES.keys()) def get_task_fixture(task_id: str) -> TaskFixture: if task_id not in TASK_FIXTURES: raise KeyError(f"Unknown task_id: {task_id}") return TASK_FIXTURES[task_id]