"""Allowlisted cancellation causes carried by asyncio's cancellation message.

Only the owner of an interruption assigns a cause. A bare CancelledError, an
HTTP status or arbitrary message cannot prove who cancelled the operation.
"""

import asyncio
from enum import StrEnum


class CancellationReason(StrEnum):
    USER_STOP = "user_stop"
    GUIDANCE_REPLACED = "guidance_replaced"
    TASK_PAUSED = "task_paused"
    TASK_REPLACED = "task_replaced"
    SUPERSEDED = "superseded"
    OUTER_TIMEOUT = "outer_timeout"
    UNKNOWN = "unknown"


def cancellation_reason(error: BaseException | None) -> CancellationReason:
    if isinstance(error, KeyboardInterrupt):
        return CancellationReason.USER_STOP
    if isinstance(error, asyncio.CancelledError) and error.args:
        reason = error.args[0]
        if isinstance(reason, CancellationReason):
            return reason
    return CancellationReason.UNKNOWN


def reason_from_code(value: object) -> CancellationReason:
    """Decode only our marker field, never a free-form user instruction."""
    if isinstance(value, str):
        try:
            return CancellationReason(value)
        except ValueError:
            pass
    return CancellationReason.UNKNOWN