support_ticket_env/fixtures.py

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]