pcd/tasks.py

"""Fixed hand-authored diagnostics, not representative production benchmarks."""


def object_schema(properties):
    return {
        "type": "object",
        "properties": properties,
        "required": list(properties),
        "additionalProperties": False,
    }


SUPPORT = object_schema(
    {
        "topic": {
            "type": "string",
            "enum": ["billing", "technical", "shipping"],
            "description": "Main issue",
        },
        "urgent": {
            "type": "boolean",
            "description": "True only if immediate action is explicitly requested",
        },
        "refund": {
            "type": "boolean",
            "description": "Whether the customer explicitly requests a refund",
        },
    }
)
ROUTING = object_schema(
    {
        "route": {
            "type": "string",
            "enum": ["north", "north west", "south"],
            "description": "Requested route",
        },
        "service": {
            "type": "string",
            "enum": ["express", "express plus", "standard"],
            "description": "Requested service level",
        },
        "insured": {"type": "boolean", "description": "Whether insurance is requested"},
    }
)
SENTIMENT = object_schema(
    {
        "sentiment": {
            "type": "string",
            "enum": ["positive", "negative", "neutral"],
            "description": "Sentiment of the message",
        },
        "language": {
            "type": "string",
            "enum": ["English", "French", "Spanish"],
            "description": "Language used",
        },
        "question": {
            "type": "boolean",
            "description": "Whether the message asks a question",
        },
    }
)


def diagnostic():
    groups = [
        (
            "support",
            SUPPORT,
            [
                (
                    "I was charged twice. Please refund the duplicate charge. No hurry.",
                    ["billing", False, True],
                ),
                (
                    "The server is down. Please fix it immediately. I do not want a refund.",
                    ["technical", True, False],
                ),
                (
                    "Where is my parcel? A status update next week is fine.",
                    ["shipping", False, False],
                ),
                (
                    "My delivery never arrived. Refund me today, this is urgent.",
                    ["shipping", True, True],
                ),
            ],
        ),
        (
            "routing",
            ROUTING,
            [
                (
                    "Route north west using express plus. Include insurance.",
                    ["north west", "express plus", True],
                ),
                (
                    "Send south with standard service, without insurance.",
                    ["south", "standard", False],
                ),
                (
                    "Use north and express, not express plus. No insurance.",
                    ["north", "express", False],
                ),
                (
                    "Route north west. Standard service. Insurance is required.",
                    ["north west", "standard", True],
                ),
            ],
        ),
        (
            "sentiment",
            SENTIMENT,
            [
                (
                    "I love this product. It works wonderfully!",
                    ["positive", "English", False],
                ),
                (
                    "Ce produit est horrible. Pouvez-vous le remplacer ?",
                    ["negative", "French", True],
                ),
                ("¿Cuál es el horario de apertura?", ["neutral", "Spanish", True]),
                (
                    "The office opens at nine and closes at five.",
                    ["neutral", "English", False],
                ),
            ],
        ),
    ]
    return [
        {
            "id": f"{group}-{i}",
            "schema": schema,
            "context": context,
            "expected": dict(zip(schema["properties"], values)),
        }
        for group, schema, rows in groups
        for i, (context, values) in enumerate(rows)
    ]


def audit():
    """Small post-freeze audit; never use these labels to select the prompt."""
    cases = [
        (
            SUPPORT,
            "Could you explain the monthly subscription price? This isn't urgent, and I'm not asking for money back.",
            ["billing", False, False],
        ),
        (
            SUPPORT,
            "Please refund my broken software purchase right now; it crashes whenever I open it.",
            ["technical", True, True],
        ),
        (
            ROUTING,
            "Use standard delivery to the north. Add insurance.",
            ["north", "standard", True],
        ),
        (
            ROUTING,
            "Express plus to the south, insurance declined.",
            ["south", "express plus", False],
        ),
        (SENTIMENT, "¡Excelente servicio, muchas gracias!", ["positive", "Spanish", False]),
        (SENTIMENT, "Le train part à huit heures.", ["neutral", "French", False]),
    ]
    return [
        {
            "id": f"audit-{i}",
            "schema": schema,
            "context": context,
            "expected": dict(zip(schema["properties"], values)),
        }
        for i, (schema, context, values) in enumerate(cases)
    ]


def stress():
    cases = []
    for n in (4, 12, 28):
        properties = {
            f"flag_{i:02}": {
                "type": "boolean",
                "description": f"Value of flag_{i:02}, true iff enabled",
            }
            for i in range(n)
        }
        expected = {name: i % 3 != 0 for i, name in enumerate(properties)}
        context = "Configuration report.\n" + "\n".join(
            f"{key}: {'enabled' if value else 'disabled'}" for key, value in expected.items()
        )
        cases.append(
            {
                "id": f"fields-{n}",
                "schema": object_schema(properties),
                "context": context,
                "expected": expected,
            }
        )
    for count in (64, 255):
        choices = [f"category-{i:03}" for i in range(count)]
        cases.append(
            {
                "id": f"enum-{count}",
                "schema": object_schema({"category": {"type": "string", "enum": choices}}),
                "context": f"The assigned category is category-{count - 7:03}. Return that exact category.",
                "expected": {"category": f"category-{count - 7:03}"},
            }
        )
    return cases